Merge branch 'main' into erik/durable-rename-fsync-metrics

.github/actionlint.yml

@@ -20,4 +20,3 @@ config-variables:
   - REMOTE_STORAGE_AZURE_REGION
   - SLACK_UPCOMING_RELEASE_CHANNEL_ID
   - DEV_AWS_OIDC_ROLE_ARN
-  - BENCHMARK_INGEST_TARGET_PROJECTID

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

@@ -221,8 +221,6 @@ runs:
         REPORT_URL: ${{ steps.generate-report.outputs.report-url }}
         COMMIT_SHA: ${{ github.event.pull_request.head.sha || github.sha }}
       with:
-        # Retry script for 5XX server errors: https://github.com/actions/github-script#retries
-        retries: 5
         script: |
           const { REPORT_URL, COMMIT_SHA } = process.env
 

.github/actions/set-docker-config-dir/action.yml

@@ -0,0 +1,36 @@
+name: "Set custom docker config directory"
+description: "Create a directory for docker config and set DOCKER_CONFIG"
+
+# Use custom DOCKER_CONFIG directory to avoid conflicts with default settings
+runs:
+  using: "composite"
+  steps:
+  - name: Show warning on GitHub-hosted runners
+    if: runner.environment == 'github-hosted'
+    shell: bash -euo pipefail {0}
+    run: |
+      # Using the following environment variables to find a path to the workflow file
+      # ${GITHUB_WORKFLOW_REF} - octocat/hello-world/.github/workflows/my-workflow.yml@refs/heads/my_branch
+      # ${GITHUB_REPOSITORY}   - octocat/hello-world
+      # ${GITHUB_REF}          - refs/heads/my_branch
+      # From https://docs.github.com/en/actions/writing-workflows/choosing-what-your-workflow-does/variables
+
+      filename_with_ref=${GITHUB_WORKFLOW_REF#"$GITHUB_REPOSITORY/"}
+      filename=${filename_with_ref%"@$GITHUB_REF"}
+
+      # https://docs.github.com/en/actions/writing-workflows/choosing-what-your-workflow-does/workflow-commands-for-github-actions#setting-a-warning-message
+      title='Unnecessary usage of `.github/actions/set-docker-config-dir`'
+      message='No need to use `.github/actions/set-docker-config-dir` action on GitHub-hosted runners'
+      echo "::warning file=${filename},title=${title}::${message}"
+
+  - uses: pyTooling/Actions/with-post-step@74afc5a42a17a046c90c68cb5cfa627e5c6c5b6b # v1.0.7
+    env:
+      DOCKER_CONFIG: .docker-custom-${{ github.run_id }}-${{ github.run_attempt }}
+    with:
+      main: |
+        mkdir -p "${DOCKER_CONFIG}"
+        echo DOCKER_CONFIG=${DOCKER_CONFIG} | tee -a $GITHUB_ENV
+      post: |
+        if [ -d "${DOCKER_CONFIG}" ]; then
+          rm -r "${DOCKER_CONFIG}"
+        fi

.github/pull_request_template.md

@@ -1,3 +1,14 @@
 ## Problem
 
 ## Summary of changes
+
+## Checklist before requesting a review
+
+- [ ] I have performed a self-review of my code.
+- [ ] If it is a core feature, I have added thorough tests.
+- [ ] Do we need to implement analytics? if so did you add the relevant metrics to the dashboard?
+- [ ] If this PR requires public announcement, mark it with /release-notes label and add several sentences in this section.
+
+## Checklist before merging
+
+- [ ] Do not forget to reformat commit message to not include the above checklist

.github/workflows/_check-codestyle-python.yml

@@ -1,37 +0,0 @@
-name: Check Codestyle Python
-
-on:
-  workflow_call:
-    inputs:
-      build-tools-image:
-        description: 'build-tools image'
-        required: true
-        type: string
-
-defaults:
-  run:
-    shell: bash -euxo pipefail {0}
-
-jobs:
-  check-codestyle-python:
-    runs-on: [ self-hosted, small ]
-    container:
-      image: ${{ inputs.build-tools-image }}
-      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
-      options: --init
-
-    steps:
-      - uses: actions/checkout@v4
-
-      - uses: actions/cache@v4
-        with:
-          path: ~/.cache/pypoetry/virtualenvs
-          key: v2-${{ runner.os }}-${{ runner.arch }}-python-deps-bookworm-${{ hashFiles('poetry.lock') }}
-
-      - run: ./scripts/pysync
-
-      - run: poetry run ruff check .
-      - run: poetry run ruff format --check .
-      - run: poetry run mypy .

.github/workflows/build-build-tools-image.yml

@@ -64,7 +64,7 @@ jobs:
 
       - uses: actions/checkout@v4
 
-      - uses: neondatabase/dev-actions/set-docker-config-dir@6094485bf440001c94a94a3f9e221e81ff6b6193
+      - uses: ./.github/actions/set-docker-config-dir
       - uses: docker/setup-buildx-action@v3
         with:
           cache-binary: false

.github/workflows/build_and_test.yml

@@ -90,10 +90,35 @@ jobs:
 
   check-codestyle-python:
     needs: [ check-permissions, build-build-tools-image ]
-    uses: ./.github/workflows/_check-codestyle-python.yml
-    with:
-      build-tools-image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
-    secrets: inherit
+    runs-on: [ self-hosted, small ]
+    container:
+      image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
+      credentials:
+        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
+        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+      options: --init
+
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Cache poetry deps
+        uses: actions/cache@v4
+        with:
+          path: ~/.cache/pypoetry/virtualenvs
+          key: v2-${{ runner.os }}-${{ runner.arch }}-python-deps-bookworm-${{ hashFiles('poetry.lock') }}
+
+      - name: Install Python deps
+        run: ./scripts/pysync
+
+      - name: Run `ruff check` to ensure code format
+        run: poetry run ruff check .
+
+      - name: Run `ruff format` to ensure code format
+        run: poetry run ruff format --check .
+
+      - name: Run mypy to check types
+        run: poetry run mypy .
 
   check-codestyle-jsonnet:
     needs: [ check-permissions, build-build-tools-image ]
@@ -116,7 +141,6 @@ jobs:
   # Check that the vendor/postgres-* submodules point to the
   # corresponding REL_*_STABLE_neon branches.
   check-submodules:
-    needs: [ check-permissions ]
     runs-on: ubuntu-22.04
     steps:
       - name: Checkout
@@ -497,8 +521,6 @@ jobs:
           REPORT_URL_NEW: ${{ steps.upload-coverage-report-new.outputs.report-url }}
           COMMIT_SHA: ${{ github.event.pull_request.head.sha || github.sha }}
         with:
-          # Retry script for 5XX server errors: https://github.com/actions/github-script#retries
-          retries: 5
           script: |
             const { REPORT_URL_NEW, COMMIT_SHA } = process.env
 
@@ -530,7 +552,7 @@ jobs:
         with:
           submodules: true
 
-      - uses: neondatabase/dev-actions/set-docker-config-dir@6094485bf440001c94a94a3f9e221e81ff6b6193
+      - uses: ./.github/actions/set-docker-config-dir
       - uses: docker/setup-buildx-action@v3
         with:
           cache-binary: false
@@ -621,7 +643,7 @@ jobs:
         with:
           submodules: true
 
-      - uses: neondatabase/dev-actions/set-docker-config-dir@6094485bf440001c94a94a3f9e221e81ff6b6193
+      - uses: ./.github/actions/set-docker-config-dir
       - uses: docker/setup-buildx-action@v3
         with:
           cache-binary: false
@@ -802,7 +824,7 @@ jobs:
           curl -fL https://github.com/neondatabase/autoscaling/releases/download/$VM_BUILDER_VERSION/vm-builder -o vm-builder
           chmod +x vm-builder
 
-      - uses: neondatabase/dev-actions/set-docker-config-dir@6094485bf440001c94a94a3f9e221e81ff6b6193
+      - uses: ./.github/actions/set-docker-config-dir
       - uses: docker/login-action@v3
         with:
           username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
@@ -838,7 +860,7 @@ jobs:
     steps:
       - uses: actions/checkout@v4
 
-      - uses: neondatabase/dev-actions/set-docker-config-dir@6094485bf440001c94a94a3f9e221e81ff6b6193
+      - uses: ./.github/actions/set-docker-config-dir
       - uses: docker/login-action@v3
         with:
           username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}

.github/workflows/ingest_benchmark.yml

@@ -1,372 +0,0 @@
-name: Benchmarking
-
-on:
-  # uncomment to run on push for debugging your PR
-  # push:
-  #   branches: [ your branch ]
-  schedule:
-    # * is a special character in YAML so you have to quote this string
-    #          ┌───────────── minute (0 - 59)
-    #          │ ┌───────────── hour (0 - 23)
-    #          │ │ ┌───────────── day of the month (1 - 31)
-    #          │ │ │ ┌───────────── month (1 - 12 or JAN-DEC)
-    #          │ │ │ │ ┌───────────── day of the week (0 - 6 or SUN-SAT)
-    - cron:   '0 9 * * *' # run once a day, timezone is utc
-  workflow_dispatch: # adds ability to run this manually
-    
-defaults:
-  run:
-    shell: bash -euxo pipefail {0}
-
-concurrency:
-  # Allow only one workflow globally because we need dedicated resources which only exist once
-  group: ingest-bench-workflow
-  cancel-in-progress: true
-
-jobs:
-  ingest:
-    strategy:
-      matrix:
-        target_project: [new_empty_project, large_existing_project]  
-    permissions:
-      contents: write
-      statuses: write
-      id-token: write # aws-actions/configure-aws-credentials
-    env:
-      PG_CONFIG: /tmp/neon/pg_install/v16/bin/pg_config
-      PSQL: /tmp/neon/pg_install/v16/bin/psql
-      PG_16_LIB_PATH: /tmp/neon/pg_install/v16/lib
-      PGCOPYDB: /pgcopydb/bin/pgcopydb
-      PGCOPYDB_LIB_PATH: /pgcopydb/lib
-    runs-on: [ self-hosted, us-east-2, x64 ]
-    container:
-      image: neondatabase/build-tools:pinned-bookworm
-      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
-      options: --init
-    timeout-minutes: 1440
-
-    steps:
-    - uses: actions/checkout@v4
-
-    - name: Configure AWS credentials # necessary to download artefacts
-      uses: aws-actions/configure-aws-credentials@v4
-      with:
-        aws-region: eu-central-1
-        role-to-assume: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
-        role-duration-seconds: 18000 # 5 hours is currently max associated with IAM role 
-
-    - name: Download Neon artifact
-      uses: ./.github/actions/download
-      with:
-        name: neon-${{ runner.os }}-${{ runner.arch }}-release-artifact
-        path: /tmp/neon/
-        prefix: latest
-
-    - name: Create Neon 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: 16
-        compute_units: '[7, 7]' # we want to test large compute here to avoid compute-side bottleneck
-        api_key: ${{ secrets.NEON_STAGING_API_KEY }}
-
-    - name: Initialize Neon project and retrieve current backpressure seconds
-      if: ${{ matrix.target_project == 'new_empty_project' }}
-      env:
-          NEW_PROJECT_CONNSTR: ${{ steps.create-neon-project-ingest-target.outputs.dsn }}
-          NEW_PROJECT_ID: ${{ steps.create-neon-project-ingest-target.outputs.project_id }}
-      run: |
-        echo "Initializing Neon project with project_id: ${NEW_PROJECT_ID}"
-        export LD_LIBRARY_PATH=${PG_16_LIB_PATH}
-        ${PSQL} "${NEW_PROJECT_CONNSTR}" -c "CREATE EXTENSION IF NOT EXISTS neon; CREATE EXTENSION IF NOT EXISTS neon_utils;"
-        BACKPRESSURE_TIME_BEFORE_INGEST=$(${PSQL} "${NEW_PROJECT_CONNSTR}" -t -c "select backpressure_throttling_time()/1000000;")
-        echo "BACKPRESSURE_TIME_BEFORE_INGEST=${BACKPRESSURE_TIME_BEFORE_INGEST}" >> $GITHUB_ENV
-        echo "NEW_PROJECT_CONNSTR=${NEW_PROJECT_CONNSTR}" >> $GITHUB_ENV
-
-    - name: Create Neon Branch for large tenant
-      if: ${{ matrix.target_project == 'large_existing_project' }}
-      id: create-neon-branch-ingest-target
-      uses: ./.github/actions/neon-branch-create
-      with:
-        project_id: ${{ vars.BENCHMARK_INGEST_TARGET_PROJECTID }}
-        api_key: ${{ secrets.NEON_STAGING_API_KEY }}
-
-    - name: Initialize Neon project and retrieve current backpressure seconds
-      if: ${{ matrix.target_project == 'large_existing_project' }}
-      env:
-          NEW_PROJECT_CONNSTR: ${{ steps.create-neon-branch-ingest-target.outputs.dsn }}
-          NEW_BRANCH_ID: ${{ steps.create-neon-branch-ingest-target.outputs.branch_id }}
-      run: |
-        echo "Initializing Neon branch with branch_id: ${NEW_BRANCH_ID}"
-        export LD_LIBRARY_PATH=${PG_16_LIB_PATH}
-        # Extract the part before the database name
-        base_connstr="${NEW_PROJECT_CONNSTR%/*}"
-        # Extract the query parameters (if any) after the database name
-        query_params="${NEW_PROJECT_CONNSTR#*\?}"
-        # Reconstruct the new connection string
-        if [ "$query_params" != "$NEW_PROJECT_CONNSTR" ]; then
-          new_connstr="${base_connstr}/neondb?${query_params}"
-        else
-          new_connstr="${base_connstr}/neondb"
-        fi
-        ${PSQL} "${new_connstr}" -c "drop database ludicrous;"
-        ${PSQL} "${new_connstr}" -c "CREATE DATABASE ludicrous;"
-        if [ "$query_params" != "$NEW_PROJECT_CONNSTR" ]; then
-          NEW_PROJECT_CONNSTR="${base_connstr}/ludicrous?${query_params}"
-        else
-          NEW_PROJECT_CONNSTR="${base_connstr}/ludicrous"
-        fi
-        ${PSQL} "${NEW_PROJECT_CONNSTR}" -c "CREATE EXTENSION IF NOT EXISTS neon; CREATE EXTENSION IF NOT EXISTS neon_utils;"
-        BACKPRESSURE_TIME_BEFORE_INGEST=$(${PSQL} "${NEW_PROJECT_CONNSTR}" -t -c "select backpressure_throttling_time()/1000000;")
-        echo "BACKPRESSURE_TIME_BEFORE_INGEST=${BACKPRESSURE_TIME_BEFORE_INGEST}" >> $GITHUB_ENV
-        echo "NEW_PROJECT_CONNSTR=${NEW_PROJECT_CONNSTR}" >> $GITHUB_ENV
-      
-        
-    - name: Create pgcopydb filter file
-      run: |
-        cat << EOF > /tmp/pgcopydb_filter.txt
-          [include-only-table]
-          public.events
-          public.emails
-          public.email_transmissions
-          public.payments
-          public.editions
-          public.edition_modules
-          public.sp_content
-          public.email_broadcasts
-          public.user_collections
-          public.devices
-          public.user_accounts
-          public.lessons
-          public.lesson_users
-          public.payment_methods
-          public.orders
-          public.course_emails
-          public.modules
-          public.users
-          public.module_users
-          public.courses
-          public.payment_gateway_keys
-          public.accounts
-          public.roles
-          public.payment_gateways
-          public.management
-          public.event_names
-        EOF
-
-    - name: Invoke pgcopydb
-      env:
-          BENCHMARK_INGEST_SOURCE_CONNSTR: ${{ secrets.BENCHMARK_INGEST_SOURCE_CONNSTR }}
-      run: |
-        export LD_LIBRARY_PATH=${PGCOPYDB_LIB_PATH}:${PG_16_LIB_PATH}
-        export PGCOPYDB_SOURCE_PGURI="${BENCHMARK_INGEST_SOURCE_CONNSTR}"
-        export PGCOPYDB_TARGET_PGURI="${NEW_PROJECT_CONNSTR}"
-        export PGOPTIONS="-c maintenance_work_mem=8388608 -c max_parallel_maintenance_workers=7"
-        ${PG_CONFIG} --bindir
-        ${PGCOPYDB} --version
-        ${PGCOPYDB} clone --skip-vacuum  --no-owner --no-acl --skip-db-properties --table-jobs 4 \
-          --index-jobs 4 --restore-jobs 4 --split-tables-larger-than 10GB --skip-extensions \
-          --use-copy-binary --filters /tmp/pgcopydb_filter.txt 2>&1 | tee /tmp/pgcopydb_${{ matrix.target_project }}.log
-
-    # create dummy pgcopydb log to test parsing
-    # - name: create dummy log for parser test
-    #   run: |
-    #     cat << EOF > /tmp/pgcopydb_${{ matrix.target_project }}.log
-    #     2024-11-04 18:00:53.433 500861 INFO   main.c:136                Running pgcopydb version 0.17.10.g8361a93 from "/usr/lib/postgresql/17/bin/pgcopydb"
-    #     2024-11-04 18:00:53.434 500861 INFO   cli_common.c:1225         [SOURCE] Copying database from "postgres://neondb_owner@ep-bitter-shape-w2c1ir0a.us-east-2.aws.neon.build/neondb?sslmode=require&keepalives=1&keepalives_idle=10&keepalives_interval=10&keepalives_count=60"
-    #     2024-11-04 18:00:53.434 500861 INFO   cli_common.c:1226         [TARGET] Copying database into "postgres://neondb_owner@ep-icy-union-w25qd5pj.us-east-2.aws.neon.build/ludicrous?sslmode=require&keepalives=1&keepalives_idle=10&keepalives_interval=10&keepalives_count=60"
-    #     2024-11-04 18:00:53.442 500861 INFO   copydb.c:105              Using work dir "/tmp/pgcopydb"
-    #     2024-11-04 18:00:53.541 500861 INFO   snapshot.c:107            Exported snapshot "00000008-00000033-1" from the source database
-    #     2024-11-04 18:00:53.556 500865 INFO   cli_clone_follow.c:543    STEP 1: fetch source database tables, indexes, and sequences
-    #     2024-11-04 18:00:54.570 500865 INFO   copydb_schema.c:716       Splitting source candidate tables larger than 10 GB
-    #     2024-11-04 18:00:54.570 500865 INFO   copydb_schema.c:829       Table public.events is 96 GB large which is larger than --split-tables-larger-than 10 GB, and does not have a unique column of type integer: splitting by CTID
-    #     2024-11-04 18:01:05.538 500865 INFO   copydb_schema.c:905       Table public.events is 96 GB large, 10 COPY processes will be used, partitioning on ctid.
-    #     2024-11-04 18:01:05.564 500865 INFO   copydb_schema.c:905       Table public.email_transmissions is 27 GB large, 4 COPY processes will be used, partitioning on id.
-    #     2024-11-04 18:01:05.584 500865 INFO   copydb_schema.c:905       Table public.lessons is 25 GB large, 4 COPY processes will be used, partitioning on id.
-    #     2024-11-04 18:01:05.605 500865 INFO   copydb_schema.c:905       Table public.lesson_users is 16 GB large, 3 COPY processes will be used, partitioning on id.
-    #     2024-11-04 18:01:05.605 500865 INFO   copydb_schema.c:761       Fetched information for 26 tables (including 4 tables split in 21 partitions total), with an estimated total of 907 million tuples and 175 GB on-disk
-    #     2024-11-04 18:01:05.687 500865 INFO   copydb_schema.c:968       Fetched information for 57 indexes (supporting 25 constraints)
-    #     2024-11-04 18:01:05.753 500865 INFO   sequences.c:78            Fetching information for 24 sequences
-    #     2024-11-04 18:01:05.903 500865 INFO   copydb_schema.c:1122      Fetched information for 4 extensions
-    #     2024-11-04 18:01:06.178 500865 INFO   copydb_schema.c:1538      Found 0 indexes (supporting 0 constraints) in the target database
-    #     2024-11-04 18:01:06.184 500865 INFO   cli_clone_follow.c:584    STEP 2: dump the source database schema (pre/post data)
-    #     2024-11-04 18:01:06.186 500865 INFO   pgcmd.c:468                /usr/lib/postgresql/16/bin/pg_dump -Fc --snapshot 00000008-00000033-1 --section=pre-data --section=post-data --file /tmp/pgcopydb/schema/schema.dump 'postgres://neondb_owner@ep-bitter-shape-w2c1ir0a.us-east-2.aws.neon.build/neondb?sslmode=require&keepalives=1&keepalives_idle=10&keepalives_interval=10&keepalives_count=60'
-    #     2024-11-04 18:01:06.952 500865 INFO   cli_clone_follow.c:592    STEP 3: restore the pre-data section to the target database
-    #     2024-11-04 18:01:07.004 500865 INFO   pgcmd.c:1001               /usr/lib/postgresql/16/bin/pg_restore --dbname 'postgres://neondb_owner@ep-icy-union-w25qd5pj.us-east-2.aws.neon.build/ludicrous?sslmode=require&keepalives=1&keepalives_idle=10&keepalives_interval=10&keepalives_count=60' --section pre-data --jobs 4 --no-owner --no-acl --use-list /tmp/pgcopydb/schema/pre-filtered.list /tmp/pgcopydb/schema/schema.dump
-    #     2024-11-04 18:01:07.438 500874 INFO   table-data.c:656          STEP 4: starting 4 table-data COPY processes
-    #     2024-11-04 18:01:07.451 500877 INFO   vacuum.c:139              STEP 8: skipping VACUUM jobs per --skip-vacuum
-    #     2024-11-04 18:01:07.457 500875 INFO   indexes.c:182             STEP 6: starting 4 CREATE INDEX processes
-    #     2024-11-04 18:01:07.457 500875 INFO   indexes.c:183             STEP 7: constraints are built by the CREATE INDEX processes
-    #     2024-11-04 18:01:07.507 500865 INFO   blobs.c:74                Skipping large objects: none found.
-    #     2024-11-04 18:01:07.509 500865 INFO   sequences.c:194           STEP 9: reset sequences values
-    #     2024-11-04 18:01:07.510 500886 INFO   sequences.c:290           Set sequences values on the target database
-    #     2024-11-04 20:49:00.587 500865 INFO   cli_clone_follow.c:608    STEP 10: restore the post-data section to the target database
-    #     2024-11-04 20:49:00.600 500865 INFO   pgcmd.c:1001               /usr/lib/postgresql/16/bin/pg_restore --dbname 'postgres://neondb_owner@ep-icy-union-w25qd5pj.us-east-2.aws.neon.build/ludicrous?sslmode=require&keepalives=1&keepalives_idle=10&keepalives_interval=10&keepalives_count=60' --section post-data --jobs 4 --no-owner --no-acl --use-list /tmp/pgcopydb/schema/post-filtered.list /tmp/pgcopydb/schema/schema.dump
-    #     2024-11-05 10:50:58.508 500865 INFO   cli_clone_follow.c:639    All step are now done, 16h49m elapsed
-    #     2024-11-05 10:50:58.508 500865 INFO   summary.c:3155            Printing summary for 26 tables and 57 indexes
-
-    #       OID | Schema |                 Name | Parts | copy duration | transmitted bytes | indexes | create index duration 
-    #     ------+--------+----------------------+-------+---------------+-------------------+---------+----------------------
-    #     24654 | public |               events |    10 |         1d11h |            878 GB |       1 |                 1h41m
-    #     24623 | public |  email_transmissions |     4 |         4h46m |             99 GB |       3 |                 2h04m
-    #     24665 | public |              lessons |     4 |         4h42m |            161 GB |       4 |                 1m11s
-    #     24661 | public |         lesson_users |     3 |         2h46m |             49 GB |       3 |                39m35s
-    #     24631 | public |               emails |     1 |        34m07s |             10 GB |       2 |                   17s
-    #     24739 | public |             payments |     1 |         5m47s |           1848 MB |       4 |                 4m40s
-    #     24681 | public |         module_users |     1 |         4m57s |           1610 MB |       3 |                 1m50s
-    #     24694 | public |               orders |     1 |         2m50s |            835 MB |       3 |                 1m05s
-    #     24597 | public |              devices |     1 |         1m45s |            498 MB |       2 |                   40s
-    #     24723 | public |      payment_methods |     1 |         1m24s |            548 MB |       2 |                   31s
-    #     24765 | public |     user_collections |     1 |         2m17s |           1005 MB |       2 |                 968ms
-    #     24774 | public |                users |     1 |           52s |            291 MB |       4 |                   27s
-    #     24760 | public |        user_accounts |     1 |           16s |            172 MB |       3 |                   16s
-    #     24606 | public |      edition_modules |     1 |         8s983 |             46 MB |       3 |                 4s749
-    #     24583 | public |        course_emails |     1 |         8s526 |             26 MB |       2 |                 996ms
-    #     24685 | public |              modules |     1 |         1s592 |             21 MB |       3 |                 1s696
-    #     24610 | public |             editions |     1 |         2s199 |           7483 kB |       2 |                 1s032
-    #     24755 | public |           sp_content |     1 |         1s555 |           4177 kB |       0 |                   0ms
-    #     24619 | public |     email_broadcasts |     1 |         744ms |           2645 kB |       2 |                 677ms
-    #     24590 | public |              courses |     1 |         387ms |           1540 kB |       2 |                 367ms
-    #     24704 | public | payment_gateway_keys |     1 |         1s972 |            164 kB |       2 |                  27ms
-    #     24576 | public |             accounts |     1 |          58ms |             24 kB |       1 |                  14ms
-    #     24647 | public |          event_names |     1 |          32ms |             397 B |       1 |                   8ms
-    #     24716 | public |     payment_gateways |     1 |         1s675 |             117 B |       1 |                  11ms
-    #     24748 | public |                roles |     1 |          71ms |             173 B |       1 |                   8ms
-    #     24676 | public |           management |     1 |          33ms |              40 B |       1 |                  19ms
-
-
-    #                                                   Step   Connection    Duration    Transfer   Concurrency
-    #     --------------------------------------------------   ----------  ----------  ----------  ------------
-    #       Catalog Queries (table ordering, filtering, etc)       source         12s                         1
-    #                                             Dump Schema       source       765ms                         1
-    #                                         Prepare Schema       target       466ms                         1
-    #           COPY, INDEX, CONSTRAINTS, VACUUM (wall clock)         both       2h47m                        12
-    #                                       COPY (cumulative)         both       7h46m     1225 GB             4
-    #                               CREATE INDEX (cumulative)       target       4h36m                         4
-    #                               CONSTRAINTS (cumulative)       target       8s493                         4
-    #                                     VACUUM (cumulative)       target         0ms                         4
-    #                                         Reset Sequences         both        60ms                         1
-    #                             Large Objects (cumulative)       (null)         0ms                         0
-    #                                         Finalize Schema         both      14h01m                         4
-    #     --------------------------------------------------   ----------  ----------  ----------  ------------
-    #                               Total Wall Clock Duration         both      16h49m                        20
-
-
-    #     EOF
-
-
-    - name: show tables sizes and retrieve current backpressure seconds
-      run: |
-        export LD_LIBRARY_PATH=${PG_16_LIB_PATH}
-        ${PSQL} "${NEW_PROJECT_CONNSTR}" -c "\dt+"
-        BACKPRESSURE_TIME_AFTER_INGEST=$(${PSQL} "${NEW_PROJECT_CONNSTR}" -t -c "select backpressure_throttling_time()/1000000;")
-        echo "BACKPRESSURE_TIME_AFTER_INGEST=${BACKPRESSURE_TIME_AFTER_INGEST}" >> $GITHUB_ENV
-
-    - name: Parse pgcopydb log and report performance metrics
-      env:
-        PERF_TEST_RESULT_CONNSTR: ${{ secrets.PERF_TEST_RESULT_CONNSTR }}
-      run: |
-        export LD_LIBRARY_PATH=${PG_16_LIB_PATH}
-
-        # Define the log file path
-        LOG_FILE="/tmp/pgcopydb_${{ matrix.target_project }}.log"
-        
-        # Get the current git commit hash
-        git config --global --add safe.directory /__w/neon/neon
-        COMMIT_HASH=$(git rev-parse --short HEAD)
-        
-        # Define the platform and test suite
-        PLATFORM="pg16-${{ matrix.target_project }}-us-east-2-staging"
-        SUIT="pgcopydb_ingest_bench"
-        
-        # Function to convert time (e.g., "2h47m", "4h36m", "118ms", "8s493") to seconds
-        convert_to_seconds() {
-          local duration=$1
-          local total_seconds=0
-    
-          # Check for hours (h)
-          if [[ "$duration" =~ ([0-9]+)h ]]; then
-            total_seconds=$((total_seconds + ${BASH_REMATCH[1]#0} * 3600))
-          fi
-    
-          # Check for seconds (s)
-          if [[ "$duration" =~ ([0-9]+)s ]]; then
-            total_seconds=$((total_seconds + ${BASH_REMATCH[1]#0}))
-          fi
-    
-          # Check for milliseconds (ms) (if applicable)
-          if [[ "$duration" =~ ([0-9]+)ms ]]; then
-            total_seconds=$((total_seconds + ${BASH_REMATCH[1]#0} / 1000))
-            duration=${duration/${BASH_REMATCH[0]}/} # need to remove it to avoid double counting with m 
-          fi
-
-          # Check for minutes (m) - must be checked after ms because m is contained in ms
-          if [[ "$duration" =~ ([0-9]+)m ]]; then
-            total_seconds=$((total_seconds + ${BASH_REMATCH[1]#0} * 60))
-          fi
-    
-          echo $total_seconds
-        }
-
-        # Calculate the backpressure difference in seconds
-        BACKPRESSURE_TIME_DIFF=$(awk "BEGIN {print $BACKPRESSURE_TIME_AFTER_INGEST - $BACKPRESSURE_TIME_BEFORE_INGEST}")
-
-        # Insert the backpressure time difference into the performance database
-        if [ -n "$BACKPRESSURE_TIME_DIFF" ]; then
-          PSQL_CMD="${PSQL} \"${PERF_TEST_RESULT_CONNSTR}\" -c \"
-          INSERT INTO public.perf_test_results (suit, revision, platform, metric_name, metric_value, metric_unit, metric_report_type, recorded_at_timestamp)
-          VALUES ('${SUIT}', '${COMMIT_HASH}', '${PLATFORM}', 'backpressure_time', ${BACKPRESSURE_TIME_DIFF}, 'seconds', 'lower_is_better', now());
-          \""
-          echo "Inserting backpressure time difference: ${BACKPRESSURE_TIME_DIFF} seconds"
-          eval $PSQL_CMD
-        fi
-
-        # Extract and process log lines
-        while IFS= read -r line; do
-          METRIC_NAME=""
-          # Match each desired line and extract the relevant information
-          if [[ "$line" =~ COPY,\ INDEX,\ CONSTRAINTS,\ VACUUM.* ]]; then
-            METRIC_NAME="COPY, INDEX, CONSTRAINTS, VACUUM (wall clock)"
-          elif [[ "$line" =~ COPY\ \(cumulative\).* ]]; then
-            METRIC_NAME="COPY (cumulative)"
-          elif [[ "$line" =~ CREATE\ INDEX\ \(cumulative\).* ]]; then
-            METRIC_NAME="CREATE INDEX (cumulative)"
-          elif [[ "$line" =~ CONSTRAINTS\ \(cumulative\).* ]]; then
-            METRIC_NAME="CONSTRAINTS (cumulative)"
-          elif [[ "$line" =~ Finalize\ Schema.* ]]; then
-            METRIC_NAME="Finalize Schema"
-          elif [[ "$line" =~ Total\ Wall\ Clock\ Duration.* ]]; then
-            METRIC_NAME="Total Wall Clock Duration"
-          fi
-          
-          # If a metric was matched, insert it into the performance database
-          if [ -n "$METRIC_NAME" ]; then
-            DURATION=$(echo "$line" | grep -oP '\d+h\d+m|\d+s|\d+ms|\d{1,2}h\d{1,2}m|\d+\.\d+s' | head -n 1)
-            METRIC_VALUE=$(convert_to_seconds "$DURATION")
-            PSQL_CMD="${PSQL} \"${PERF_TEST_RESULT_CONNSTR}\" -c \"
-            INSERT INTO public.perf_test_results (suit, revision, platform, metric_name, metric_value, metric_unit, metric_report_type, recorded_at_timestamp)
-            VALUES ('${SUIT}', '${COMMIT_HASH}', '${PLATFORM}', '${METRIC_NAME}', ${METRIC_VALUE}, 'seconds', 'lower_is_better', now());
-            \""
-            echo "Inserting ${METRIC_NAME} with value ${METRIC_VALUE} seconds"
-            eval $PSQL_CMD
-          fi
-        done < "$LOG_FILE"
-      
-    - name: Delete Neon 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 }}
-        api_key: ${{ secrets.NEON_STAGING_API_KEY }}
-
-    - name: Delete Neon Branch for large tenant
-      if: ${{ always() && matrix.target_project == 'large_existing_project' }}
-      uses: ./.github/actions/neon-branch-delete
-      with:
-        project_id: ${{ vars.BENCHMARK_INGEST_TARGET_PROJECTID }}
-        branch_id: ${{ steps.create-neon-branch-ingest-target.outputs.branch_id }}
-        api_key: ${{ secrets.NEON_STAGING_API_KEY }}

.github/workflows/neon_extra_builds.yml

@@ -201,8 +201,6 @@ jobs:
           REPORT_URL: ${{ steps.upload-stats.outputs.report-url }}
           SHA: ${{ github.event.pull_request.head.sha || github.sha }}
         with:
-          # Retry script for 5XX server errors: https://github.com/actions/github-script#retries
-          retries: 5
           script: |
             const { REPORT_URL, SHA } = process.env
 

.github/workflows/pre-merge-checks.yml

@@ -1,94 +0,0 @@
-name: Pre-merge checks
-
-on:
-  merge_group:
-    branches:
-      - main
-
-defaults:
-  run:
-    shell: bash -euxo pipefail {0}
-
-# No permission for GITHUB_TOKEN by default; the **minimal required** set of permissions should be granted in each job.
-permissions: {}
-
-jobs:
-  get-changed-files:
-    runs-on: ubuntu-22.04
-    outputs:
-      python-changed: ${{ steps.python-src.outputs.any_changed }}
-    steps:
-      - uses: actions/checkout@v4
-      - uses: tj-actions/changed-files@4edd678ac3f81e2dc578756871e4d00c19191daf # v45.0.4
-        id: python-src
-        with:
-          files: |
-            .github/workflows/pre-merge-checks.yml
-            **/**.py
-            poetry.lock
-            pyproject.toml
-
-      - name: PRINT ALL CHANGED FILES FOR DEBUG PURPOSES
-        env:
-          PYTHON_CHANGED_FILES: ${{ steps.python-src.outputs.all_changed_files }}
-        run: |
-          echo "${PYTHON_CHANGED_FILES}"
-
-  check-build-tools-image:
-    if: needs.get-changed-files.outputs.python-changed == 'true'
-    needs: [ get-changed-files ]
-    uses: ./.github/workflows/check-build-tools-image.yml
-
-  build-build-tools-image:
-    needs: [ check-build-tools-image ]
-    uses: ./.github/workflows/build-build-tools-image.yml
-    with:
-      image-tag: ${{ needs.check-build-tools-image.outputs.image-tag }}
-    secrets: inherit
-
-  check-codestyle-python:
-    if: needs.get-changed-files.outputs.python-changed == 'true'
-    needs: [ get-changed-files, build-build-tools-image ]
-    uses: ./.github/workflows/_check-codestyle-python.yml
-    with:
-      build-tools-image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
-    secrets: inherit
-
-  # To get items from the merge queue merged into main we need to satisfy "Status checks that are required".
-  # Currently we require 2 jobs (checks with exact name):
-  # - conclusion
-  # - neon-cloud-e2e
-  conclusion:
-    if: always()
-    permissions:
-      statuses: write # for `github.repos.createCommitStatus(...)`
-    needs:
-      - get-changed-files
-      - check-codestyle-python
-    runs-on: ubuntu-22.04
-    steps:
-      - name: Create fake `neon-cloud-e2e` check
-        uses: actions/github-script@v7
-        with:
-          # Retry script for 5XX server errors: https://github.com/actions/github-script#retries
-          retries: 5
-          script: |
-            const { repo, owner } = context.repo;
-            const targetUrl = `${context.serverUrl}/${owner}/${repo}/actions/runs/${context.runId}`;
-
-            await github.rest.repos.createCommitStatus({
-              owner: owner,
-              repo: repo,
-              sha: context.sha,
-              context: `neon-cloud-e2e`,
-              state: `success`,
-              target_url: targetUrl,
-              description: `fake check for merge queue`,
-            });
-
-      - name: Fail the job if any of the dependencies do not succeed or skipped
-        run: exit 1
-        if: |
-          (contains(needs.check-codestyle-python.result, 'skipped') && needs.get-changed-files.outputs.python-changed == 'true')
-          || contains(needs.*.result, 'failure')
-          || contains(needs.*.result, 'cancelled')

.github/workflows/report-workflow-stats-batch.yml

@@ -1,29 +0,0 @@
-name: Report Workflow Stats Batch
-
-on:
-  schedule:
-    - cron: '*/15 * * * *'
-    - cron: '25 0 * * *'
-
-jobs:
-  gh-workflow-stats-batch:
-    name: GitHub Workflow Stats Batch
-    runs-on: ubuntu-22.04
-    permissions:
-      actions: read
-    steps:
-    - name: Export Workflow Run for the past 2 hours
-      uses: neondatabase/gh-workflow-stats-action@v0.2.1
-      with:
-        db_uri: ${{ secrets.GH_REPORT_STATS_DB_RW_CONNSTR }}
-        db_table: "gh_workflow_stats_batch_neon"
-        gh_token: ${{ secrets.GITHUB_TOKEN }}
-        duration: '2h'
-    - name: Export Workflow Run for the past 24 hours
-      if: github.event.schedule == '25 0 * * *'
-      uses: neondatabase/gh-workflow-stats-action@v0.2.1
-      with:
-        db_uri: ${{ secrets.GH_REPORT_STATS_DB_RW_CONNSTR }}
-        db_table: "gh_workflow_stats_batch_neon"
-        gh_token: ${{ secrets.GITHUB_TOKEN }}
-        duration: '24h'

.github/workflows/report-workflow-stats.yml

@@ -23,7 +23,6 @@ on:
     - Test Postgres client libraries
     - Trigger E2E Tests
     - cleanup caches by a branch
-    - Pre-merge checks
     types: [completed]
 
 jobs:

Cargo.lock

@@ -994,9 +994,9 @@ checksum = "14c189c53d098945499cdfa7ecc63567cf3886b3332b312a5b4585d8d3a6a610"
 
 [[package]]
 name = "bytes"
-version = "1.8.0"
+version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9ac0150caa2ae65ca5bd83f25c7de183dea78d4d366469f148435e2acfbad0da"
+checksum = "a2bd12c1caf447e69cd4528f47f94d203fd2582878ecb9e9465484c4148a8223"
 dependencies = [
  "serde",
 ]
@@ -1229,15 +1229,12 @@ dependencies = [
  "flate2",
  "futures",
  "hyper 0.14.30",
- "metrics",
  "nix 0.27.1",
  "notify",
  "num_cpus",
- "once_cell",
  "opentelemetry",
  "opentelemetry_sdk",
  "postgres",
- "prometheus",
  "regex",
  "remote_storage",
  "reqwest 0.12.4",
@@ -4746,7 +4743,6 @@ dependencies = [
  "percent-encoding",
  "pin-project-lite",
  "rustls 0.22.4",
- "rustls-native-certs 0.7.0",
  "rustls-pemfile 2.1.1",
  "rustls-pki-types",
  "serde",

build-tools.Dockerfile

@@ -1,66 +1,12 @@
 ARG DEBIAN_VERSION=bullseye
 
-FROM debian:bookworm-slim AS pgcopydb_builder
-ARG DEBIAN_VERSION
-
-RUN if [ "${DEBIAN_VERSION}" = "bookworm" ]; then \
-        set -e && \
-        apt update && \
-        apt install -y --no-install-recommends \
-        ca-certificates wget gpg && \
-        wget -qO - https://www.postgresql.org/media/keys/ACCC4CF8.asc | gpg --dearmor -o /usr/share/keyrings/postgresql-keyring.gpg && \
-        echo "deb [signed-by=/usr/share/keyrings/postgresql-keyring.gpg] http://apt.postgresql.org/pub/repos/apt bookworm-pgdg main" > /etc/apt/sources.list.d/pgdg.list && \
-        apt-get update && \
-        apt install -y --no-install-recommends \
-        build-essential \
-        autotools-dev \
-        libedit-dev \
-        libgc-dev \
-        libpam0g-dev \
-        libreadline-dev \
-        libselinux1-dev \
-        libxslt1-dev \
-        libssl-dev \
-        libkrb5-dev \
-        zlib1g-dev \
-        liblz4-dev \
-        libpq5 \
-        libpq-dev \
-        libzstd-dev \
-        postgresql-16 \
-        postgresql-server-dev-16 \
-        postgresql-common  \
-        python3-sphinx && \
-        wget -O /tmp/pgcopydb.tar.gz https://github.com/dimitri/pgcopydb/archive/refs/tags/v0.17.tar.gz && \
-        mkdir /tmp/pgcopydb && \
-        tar -xzf /tmp/pgcopydb.tar.gz -C /tmp/pgcopydb --strip-components=1 && \
-        cd /tmp/pgcopydb && \
-        make -s clean && \
-        make -s -j12 install && \
-        libpq_path=$(find /lib /usr/lib -name "libpq.so.5" | head -n 1) && \
-        mkdir -p /pgcopydb/lib && \
-        cp "$libpq_path" /pgcopydb/lib/; \
-    else \
-        # copy command below will fail if we don't have dummy files, so we create them for other debian versions
-        mkdir -p /usr/lib/postgresql/16/bin && touch /usr/lib/postgresql/16/bin/pgcopydb && \
-        mkdir -p mkdir -p /pgcopydb/lib && touch /pgcopydb/lib/libpq.so.5; \
-    fi
-
-FROM debian:${DEBIAN_VERSION}-slim AS build_tools
+FROM debian:${DEBIAN_VERSION}-slim
 ARG DEBIAN_VERSION
 
 # Add nonroot user
 RUN useradd -ms /bin/bash nonroot -b /home
 SHELL ["/bin/bash", "-c"]
 
-RUN mkdir -p /pgcopydb/bin && \
-    mkdir -p /pgcopydb/lib && \
-    chmod -R 755 /pgcopydb && \
-    chown -R nonroot:nonroot /pgcopydb
-        
-COPY --from=pgcopydb_builder /usr/lib/postgresql/16/bin/pgcopydb /pgcopydb/bin/pgcopydb 
-COPY --from=pgcopydb_builder /pgcopydb/lib/libpq.so.5 /pgcopydb/lib/libpq.so.5 
-
 # System deps
 #
 # 'gdb' is included so that we get backtraces of core dumps produced in
@@ -92,7 +38,7 @@ RUN set -e \
         libseccomp-dev \
         libsqlite3-dev \
         libssl-dev \
-        $([[ "${DEBIAN_VERSION}" = "bullseye" ]] && echo libstdc++-10-dev || echo libstdc++-11-dev) \
+        $([[ "${DEBIAN_VERSION}" = "bullseye" ]] && libstdc++-10-dev || libstdc++-11-dev) \
         libtool \
         libxml2-dev \
         libxmlsec1-dev \
@@ -289,13 +235,7 @@ RUN whoami \
     && cargo --version --verbose \
     && rustup --version --verbose \
     && rustc --version --verbose \
-    && clang --version 
-
-RUN if [ "${DEBIAN_VERSION}" = "bookworm" ]; then \
-    LD_LIBRARY_PATH=/pgcopydb/lib /pgcopydb/bin/pgcopydb --version; \
-else \
-    echo "pgcopydb is not available for ${DEBIAN_VERSION}"; \
-fi
+    && clang --version
 
 # Set following flag to check in Makefile if its running in Docker
 RUN touch /home/nonroot/.docker_build

compute/compute-node.Dockerfile

@@ -559,8 +559,8 @@ RUN case "${PG_VERSION}" in \
         export TIMESCALEDB_CHECKSUM=584a351c7775f0e067eaa0e7277ea88cab9077cc4c455cbbf09a5d9723dce95d \
         ;; \
       "v17") \
-        export TIMESCALEDB_VERSION=2.17.1 \
-        export TIMESCALEDB_CHECKSUM=6277cf43f5695e23dae1c5cfeba00474d730b66ed53665a84b787a6bb1a57e28 \
+        export TIMESCALEDB_VERSION=2.17.0 \
+        export TIMESCALEDB_CHECKSUM=155bf64391d3558c42f31ca0e523cfc6252921974f75298c9039ccad1c89811a \
         ;; \
     esac && \
     wget https://github.com/timescale/timescaledb/archive/refs/tags/${TIMESCALEDB_VERSION}.tar.gz -O timescaledb.tar.gz && \
@@ -624,12 +624,16 @@ FROM build-deps AS pg-cron-pg-build
 ARG PG_VERSION
 COPY --from=pg-build /usr/local/pgsql/ /usr/local/pgsql/
 
+# 1.6.4 available, supports v17
 # This is an experimental extension that we do not support on prod yet.
 # !Do not remove!
 # We set it in shared_preload_libraries and computes will fail to start if library is not found.
 ENV PATH="/usr/local/pgsql/bin/:$PATH"
-RUN wget https://github.com/citusdata/pg_cron/archive/refs/tags/v1.6.4.tar.gz -O pg_cron.tar.gz && \
-    echo "52d1850ee7beb85a4cb7185731ef4e5a90d1de216709d8988324b0d02e76af61 pg_cron.tar.gz" | sha256sum --check && \
+RUN case "${PG_VERSION}" in "v17") \
+    echo "v17 extensions are not supported yet. Quit" && exit 0;; \
+    esac && \
+    wget https://github.com/citusdata/pg_cron/archive/refs/tags/v1.6.0.tar.gz -O pg_cron.tar.gz && \
+    echo "383a627867d730222c272bfd25cd5e151c578d73f696d32910c7db8c665cc7db pg_cron.tar.gz" | sha256sum --check && \
     mkdir pg_cron-src && cd pg_cron-src && tar xzf ../pg_cron.tar.gz --strip-components=1 -C . && \
     make -j $(getconf _NPROCESSORS_ONLN) && \
     make -j $(getconf _NPROCESSORS_ONLN) install && \
@@ -1471,8 +1475,6 @@ RUN mkdir -p /etc/local_proxy && chown postgres:postgres /etc/local_proxy
 COPY --from=postgres-exporter /bin/postgres_exporter /bin/postgres_exporter
 COPY --from=sql-exporter      /bin/sql_exporter      /bin/sql_exporter
 
-COPY --chown=postgres compute/etc/postgres_exporter.yml /etc/postgres_exporter.yml
-
 COPY --from=sql_exporter_preprocessor --chmod=0644 /home/nonroot/compute/etc/sql_exporter.yml               /etc/sql_exporter.yml
 COPY --from=sql_exporter_preprocessor --chmod=0644 /home/nonroot/compute/etc/neon_collector.yml             /etc/neon_collector.yml
 COPY --from=sql_exporter_preprocessor --chmod=0644 /home/nonroot/compute/etc/sql_exporter_autoscaling.yml   /etc/sql_exporter_autoscaling.yml

compute/etc/sql_exporter/compute_backpressure_throttling_seconds.sql

@@ -1 +1 @@
-SELECT (neon.backpressure_throttling_time()::float8 / 1000000) AS throttled;
+SELECT neon.backpressure_throttling_time()::float8 / 1000 AS throttled;

compute/vm-image-spec-bookworm.yaml

@@ -26,7 +26,7 @@ commands:
   - name: postgres-exporter
     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'
+    shell: 'DATA_SOURCE_NAME="user=cloud_admin sslmode=disable dbname=postgres application_name=postgres-exporter" /bin/postgres_exporter'
   - name: sql-exporter
     user: nobody
     sysvInitAction: respawn

compute/vm-image-spec-bullseye.yaml

@@ -26,7 +26,7 @@ commands:
   - name: postgres-exporter
     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'
+    shell: 'DATA_SOURCE_NAME="user=cloud_admin sslmode=disable dbname=postgres application_name=postgres-exporter" /bin/postgres_exporter'
   - name: sql-exporter
     user: nobody
     sysvInitAction: respawn

compute_tools/Cargo.toml

@@ -18,11 +18,9 @@ clap.workspace = true
 flate2.workspace = true
 futures.workspace = true
 hyper0 = { workspace = true, features = ["full"] }
-metrics.workspace = true
 nix.workspace = true
 notify.workspace = true
 num_cpus.workspace = true
-once_cell.workspace = true
 opentelemetry.workspace = true
 opentelemetry_sdk.workspace = true
 postgres.workspace = true
@@ -41,7 +39,6 @@ tracing-subscriber.workspace = true
 tracing-utils.workspace = true
 thiserror.workspace = true
 url.workspace = true
-prometheus.workspace = true
 
 compute_api.workspace = true
 utils.workspace = true

compute_tools/src/compute.rs

@@ -364,29 +364,11 @@ impl ComputeNode {
         let pageserver_connect_micros = start_time.elapsed().as_micros() as u64;
 
         let basebackup_cmd = match lsn {
-            Lsn(0) => {
-                if spec.spec.mode != ComputeMode::Primary {
-                    format!(
-                        "basebackup {} {} --gzip --replica",
-                        spec.tenant_id, spec.timeline_id
-                    )
-                } else {
-                    format!("basebackup {} {} --gzip", spec.tenant_id, spec.timeline_id)
-                }
-            }
-            _ => {
-                if spec.spec.mode != ComputeMode::Primary {
-                    format!(
-                        "basebackup {} {} {} --gzip --replica",
-                        spec.tenant_id, spec.timeline_id, lsn
-                    )
-                } else {
-                    format!(
-                        "basebackup {} {} {} --gzip",
-                        spec.tenant_id, spec.timeline_id, lsn
-                    )
-                }
-            }
+            Lsn(0) => format!("basebackup {} {} --gzip", spec.tenant_id, spec.timeline_id),
+            _ => format!(
+                "basebackup {} {} {} --gzip",
+                spec.tenant_id, spec.timeline_id, lsn
+            ),
         };
 
         let copyreader = client.copy_out(basebackup_cmd.as_str())?;

compute_tools/src/config.rs

@@ -73,19 +73,6 @@ pub fn write_postgres_conf(
         )?;
     }
 
-    // Locales
-    if cfg!(target_os = "macos") {
-        writeln!(file, "lc_messages='C'")?;
-        writeln!(file, "lc_monetary='C'")?;
-        writeln!(file, "lc_time='C'")?;
-        writeln!(file, "lc_numeric='C'")?;
-    } else {
-        writeln!(file, "lc_messages='C.UTF-8'")?;
-        writeln!(file, "lc_monetary='C.UTF-8'")?;
-        writeln!(file, "lc_time='C.UTF-8'")?;
-        writeln!(file, "lc_numeric='C.UTF-8'")?;
-    }
-
     match spec.mode {
         ComputeMode::Primary => {}
         ComputeMode::Static(lsn) => {

compute_tools/src/http/api.rs

@@ -9,7 +9,6 @@ use crate::catalog::SchemaDumpError;
 use crate::catalog::{get_database_schema, get_dbs_and_roles};
 use crate::compute::forward_termination_signal;
 use crate::compute::{ComputeNode, ComputeState, ParsedSpec};
-use crate::installed_extensions;
 use compute_api::requests::{ConfigurationRequest, ExtensionInstallRequest, SetRoleGrantsRequest};
 use compute_api::responses::{
     ComputeStatus, ComputeStatusResponse, ExtensionInstallResult, GenericAPIError,
@@ -20,8 +19,6 @@ use anyhow::Result;
 use hyper::header::CONTENT_TYPE;
 use hyper::service::{make_service_fn, service_fn};
 use hyper::{Body, Method, Request, Response, Server, StatusCode};
-use metrics::Encoder;
-use metrics::TextEncoder;
 use tokio::task;
 use tracing::{debug, error, info, warn};
 use tracing_utils::http::OtelName;
@@ -68,28 +65,6 @@ async fn routes(req: Request<Body>, compute: &Arc<ComputeNode>) -> Response<Body
             Response::new(Body::from(serde_json::to_string(&metrics).unwrap()))
         }
 
-        // Prometheus metrics
-        (&Method::GET, "/metrics") => {
-            debug!("serving /metrics GET request");
-
-            let mut buffer = vec![];
-            let metrics = installed_extensions::collect();
-            let encoder = TextEncoder::new();
-            encoder.encode(&metrics, &mut buffer).unwrap();
-
-            match Response::builder()
-                .status(StatusCode::OK)
-                .header(CONTENT_TYPE, encoder.format_type())
-                .body(Body::from(buffer))
-            {
-                Ok(response) => response,
-                Err(err) => {
-                    let msg = format!("error handling /metrics request: {err}");
-                    error!(msg);
-                    render_json_error(&msg, StatusCode::INTERNAL_SERVER_ERROR)
-                }
-            }
-        }
         // Collect Postgres current usage insights
         (&Method::GET, "/insights") => {
             info!("serving /insights GET request");

compute_tools/src/http/openapi_spec.yaml

@@ -37,21 +37,6 @@ paths:
               schema:
                 $ref: "#/components/schemas/ComputeMetrics"
 
-  /metrics
-    get:
-      tags:
-      - Info
-      summary: Get compute node metrics in text format.
-      description: ""
-      operationId: getComputeMetrics
-      responses:
-        200:
-          description: ComputeMetrics
-          content:
-            text/plain:
-              schema:
-                type: string
-                description: Metrics in text format.
   /insights:
     get:
       tags:

compute_tools/src/installed_extensions.rs

@@ -1,5 +1,4 @@
 use compute_api::responses::{InstalledExtension, InstalledExtensions};
-use metrics::proto::MetricFamily;
 use std::collections::HashMap;
 use std::collections::HashSet;
 use tracing::info;
@@ -9,10 +8,6 @@ use anyhow::Result;
 use postgres::{Client, NoTls};
 use tokio::task;
 
-use metrics::core::Collector;
-use metrics::{register_uint_gauge_vec, UIntGaugeVec};
-use once_cell::sync::Lazy;
-
 /// We don't reuse get_existing_dbs() just for code clarity
 /// and to make database listing query here more explicit.
 ///
@@ -64,12 +59,6 @@ pub async fn get_installed_extensions(connstr: Url) -> Result<InstalledExtension
 
             for (extname, v) in extensions.iter() {
                 let version = v.to_string();
-
-                // increment the number of databases where the version of extension is installed
-                INSTALLED_EXTENSIONS
-                    .with_label_values(&[extname, &version])
-                    .inc();
-
                 extensions_map
                     .entry(extname.to_string())
                     .and_modify(|e| {
@@ -85,11 +74,9 @@ pub async fn get_installed_extensions(connstr: Url) -> Result<InstalledExtension
             }
         }
 
-        let res = InstalledExtensions {
+        Ok(InstalledExtensions {
             extensions: extensions_map.values().cloned().collect(),
-        };
-
-        Ok(res)
+        })
     })
     .await?
 }
@@ -110,18 +97,6 @@ pub fn get_installed_extensions_sync(connstr: Url) -> Result<()> {
         "[NEON_EXT_STAT] {}",
         serde_json::to_string(&result).expect("failed to serialize extensions list")
     );
+
     Ok(())
 }
-
-static INSTALLED_EXTENSIONS: Lazy<UIntGaugeVec> = Lazy::new(|| {
-    register_uint_gauge_vec!(
-        "installed_extensions",
-        "Number of databases where the version of extension is installed",
-        &["extension_name", "version"]
-    )
-    .expect("failed to define a metric")
-});
-
-pub fn collect() -> Vec<MetricFamily> {
-    INSTALLED_EXTENSIONS.collect()
-}

control_plane/src/bin/neon_local.rs

@@ -944,9 +944,6 @@ fn handle_init(args: &InitCmdArgs) -> anyhow::Result<LocalEnv> {
                         pg_auth_type: AuthType::Trust,
                         http_auth_type: AuthType::Trust,
                         other: Default::default(),
-                        // Typical developer machines use disks with slow fsync, and we don't care
-                        // about data integrity: disable disk syncs.
-                        no_sync: true,
                     }
                 })
                 .collect(),

control_plane/src/local_env.rs

@@ -225,7 +225,6 @@ pub struct PageServerConf {
     pub listen_http_addr: String,
     pub pg_auth_type: AuthType,
     pub http_auth_type: AuthType,
-    pub no_sync: bool,
 }
 
 impl Default for PageServerConf {
@@ -236,7 +235,6 @@ impl Default for PageServerConf {
             listen_http_addr: String::new(),
             pg_auth_type: AuthType::Trust,
             http_auth_type: AuthType::Trust,
-            no_sync: false,
         }
     }
 }
@@ -251,8 +249,6 @@ pub struct NeonLocalInitPageserverConf {
     pub listen_http_addr: String,
     pub pg_auth_type: AuthType,
     pub http_auth_type: AuthType,
-    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
-    pub no_sync: bool,
     #[serde(flatten)]
     pub other: HashMap<String, toml::Value>,
 }
@@ -265,7 +261,6 @@ impl From<&NeonLocalInitPageserverConf> for PageServerConf {
             listen_http_addr,
             pg_auth_type,
             http_auth_type,
-            no_sync,
             other: _,
         } = conf;
         Self {
@@ -274,7 +269,6 @@ impl From<&NeonLocalInitPageserverConf> for PageServerConf {
             listen_http_addr: listen_http_addr.clone(),
             pg_auth_type: *pg_auth_type,
             http_auth_type: *http_auth_type,
-            no_sync: *no_sync,
         }
     }
 }
@@ -575,8 +569,6 @@ impl LocalEnv {
                     listen_http_addr: String,
                     pg_auth_type: AuthType,
                     http_auth_type: AuthType,
-                    #[serde(default)]
-                    no_sync: bool,
                 }
                 let config_toml_path = dentry.path().join("pageserver.toml");
                 let config_toml: PageserverConfigTomlSubset = toml_edit::de::from_str(
@@ -599,7 +591,6 @@ impl LocalEnv {
                     listen_http_addr,
                     pg_auth_type,
                     http_auth_type,
-                    no_sync,
                 } = config_toml;
                 let IdentityTomlSubset {
                     id: identity_toml_id,
@@ -616,7 +607,6 @@ impl LocalEnv {
                     listen_http_addr,
                     pg_auth_type,
                     http_auth_type,
-                    no_sync,
                 };
                 pageservers.push(conf);
             }

control_plane/src/pageserver.rs

@@ -273,7 +273,6 @@ impl PageServerNode {
             )
         })?;
         let args = vec!["-D", datadir_path_str];
-
         background_process::start_process(
             "pageserver",
             &datadir,

docs/rfcs/038-aux-file-v2.md

@@ -91,7 +91,7 @@ generating the basebackup by scanning the `REPL_ORIGIN_KEY_PREFIX` keyspace.
 There are two places we need to read the aux files from the pageserver:
 
 * On the write path, when the compute node adds an aux file to the pageserver, we will retrieve the key from the storage, append the file to the hashed key, and write it back. The current `get` API already supports that.
-*  We use the vectored get API to retrieve all aux files during generating the basebackup. Because we need to scan a sparse keyspace, we slightly modified the vectored get path. The vectorized API used to always attempt to retrieve every single key within the requested key range, and therefore, we modified it in a way that keys within `NON_INHERITED_SPARSE_RANGE` will not trigger missing key error. Furthermore, as aux file reads usually need all layer files intersecting with that key range within the branch and cover a big keyspace, it incurs large overhead for tracking keyspaces that have not been read. Therefore, for sparse keyspaces, we [do not track](https://github.com/neondatabase/neon/pull/9631) `ummapped_keyspace`.
+*  We use the vectored get API to retrieve all aux files during generating the basebackup. Because we need to scan a sparse keyspace, we slightly modified the vectored get path. The vectorized API will attempt to retrieve every single key within the requested key range, and therefore, we modified it in a way that keys within `NON_INHERITED_SPARSE_RANGE` will not trigger missing key error.
 
 ## Compaction and Image Layer Generation
 

libs/pageserver_api/src/config.rs

@@ -64,7 +64,6 @@ pub struct ConfigToml {
     #[serde(with = "humantime_serde")]
     pub wal_redo_timeout: Duration,
     pub superuser: String,
-    pub locale: String,
     pub page_cache_size: usize,
     pub max_file_descriptors: usize,
     pub pg_distrib_dir: Option<Utf8PathBuf>,
@@ -107,8 +106,6 @@ pub struct ConfigToml {
     pub ephemeral_bytes_per_memory_kb: usize,
     pub l0_flush: Option<crate::models::L0FlushConfig>,
     pub virtual_file_io_mode: Option<crate::models::virtual_file::IoMode>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub no_sync: Option<bool>,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
@@ -277,11 +274,6 @@ pub mod defaults {
     pub const DEFAULT_WAL_REDO_TIMEOUT: &str = "60 s";
 
     pub const DEFAULT_SUPERUSER: &str = "cloud_admin";
-    pub const DEFAULT_LOCALE: &str = if cfg!(target_os = "macos") {
-        "C"
-    } else {
-        "C.UTF-8"
-    };
 
     pub const DEFAULT_PAGE_CACHE_SIZE: usize = 8192;
     pub const DEFAULT_MAX_FILE_DESCRIPTORS: usize = 100;
@@ -332,7 +324,6 @@ impl Default for ConfigToml {
             wal_redo_timeout: (humantime::parse_duration(DEFAULT_WAL_REDO_TIMEOUT)
                 .expect("cannot parse default wal redo timeout")),
             superuser: (DEFAULT_SUPERUSER.to_string()),
-            locale: DEFAULT_LOCALE.to_string(),
             page_cache_size: (DEFAULT_PAGE_CACHE_SIZE),
             max_file_descriptors: (DEFAULT_MAX_FILE_DESCRIPTORS),
             pg_distrib_dir: None, // Utf8PathBuf::from("./pg_install"), // TODO: formely, this was std::env::current_dir()
@@ -398,7 +389,6 @@ impl Default for ConfigToml {
             l0_flush: None,
             virtual_file_io_mode: None,
             tenant_config: TenantConfigToml::default(),
-            no_sync: None,
         }
     }
 }

libs/pageserver_api/src/record.rs

@@ -80,18 +80,18 @@ impl NeonWalRecord {
     }
 
     #[cfg(feature = "testing")]
-    pub fn wal_clear(s: impl AsRef<str>) -> Self {
+    pub fn wal_clear() -> Self {
         Self::Test {
-            append: s.as_ref().to_string(),
+            append: "".to_string(),
             clear: true,
             will_init: false,
         }
     }
 
     #[cfg(feature = "testing")]
-    pub fn wal_init(s: impl AsRef<str>) -> Self {
+    pub fn wal_init() -> Self {
         Self::Test {
-            append: s.as_ref().to_string(),
+            append: "".to_string(),
             clear: true,
             will_init: true,
         }

libs/utils/scripts/restore_from_wal.sh

@@ -1,4 +1,4 @@
-#!/usr/bin/env bash
+#!/bin/bash
 
 set -euxo pipefail
 
@@ -6,44 +6,9 @@ PG_BIN=$1
 WAL_PATH=$2
 DATA_DIR=$3
 PORT=$4
-PG_VERSION=$5
 SYSID=$(od -A n -j 24 -N 8 -t d8 "$WAL_PATH"/000000010000000000000002* | cut -c 3-)
-
-# The way that initdb is invoked must match how the pageserver runs initdb.
-function initdb_with_args {
-    local cmd=(
-        "$PG_BIN"/initdb
-        -E utf8
-        -U cloud_admin
-        -D "$DATA_DIR"
-        --locale 'C.UTF-8'
-        --lc-collate 'C.UTF-8'
-        --lc-ctype 'C.UTF-8'
-        --lc-messages 'C.UTF-8'
-        --lc-monetary 'C.UTF-8'
-        --lc-numeric 'C.UTF-8'
-        --lc-time 'C.UTF-8'
-        --sysid="$SYSID"
-    )
-
-    case "$PG_VERSION" in
-        14)
-            # Postgres 14 and below didn't support --locale-provider
-            ;;
-        15 | 16)
-            cmd+=(--locale-provider 'libc')
-            ;;
-        *)
-            # Postgres 17 added the builtin provider
-            cmd+=(--locale-provider 'builtin')
-            ;;
-    esac
-
-    eval env -i LD_LIBRARY_PATH="$PG_BIN"/../lib "${cmd[*]}"
-}
-
 rm -fr "$DATA_DIR"
-initdb_with_args
+env -i LD_LIBRARY_PATH="$PG_BIN"/../lib "$PG_BIN"/initdb -E utf8 -U cloud_admin -D "$DATA_DIR" --sysid="$SYSID"
 echo "port=$PORT" >> "$DATA_DIR"/postgresql.conf
 echo "shared_preload_libraries='\$libdir/neon_rmgr.so'" >> "$DATA_DIR"/postgresql.conf
 REDO_POS=0x$("$PG_BIN"/pg_controldata -D "$DATA_DIR" | grep -F "REDO location"| cut -c 42-)

libs/utils/src/auth.rs

@@ -40,11 +40,6 @@ pub enum Scope {
     /// Allows access to storage controller APIs used by the scrubber, to interrogate the state
     /// of a tenant & post scrub results.
     Scrubber,
-
-    /// This scope is used for communication with other storage controller instances.
-    /// At the time of writing, this is only used for the step down request.
-    #[serde(rename = "controller_peer")]
-    ControllerPeer,
 }
 
 /// JWT payload. See docs/authentication.md for the format

libs/utils/src/crashsafe.rs

@@ -1,4 +1,5 @@
 use std::os::fd::AsRawFd;
+use std::time::Duration;
 use std::{
     borrow::Cow,
     fs::{self, File},
@@ -123,27 +124,16 @@ pub async fn fsync_async_opt(
     Ok(())
 }
 
-/// Like postgres' durable_rename, renames a file and issues fsyncs to make it durable. After
-/// returning, both the file and rename are guaranteed to be persisted. Both paths must be on the
-/// same file system.
+/// Like postgres' durable_rename, renames file issuing fsyncs do make it
+/// durable. After return, file and rename are guaranteed to be persisted.
+/// Returns the fsync latencies, for metrics (this is kind of a kludge).
 ///
-/// Unlike postgres, it only fsyncs 1) the file to make contents durable, and 2) the directory to
-/// make the rename durable. This sequence ensures the target file will never be incomplete.
-///
-/// Postgres also:
-///
-/// * Fsyncs the target file, if it exists, before the rename, to ensure either the new or existing
-///   file survives a crash. Current callers don't need this as it should already be fsynced if
-///   durability is needed.
-///
-/// * Fsyncs the file after the rename. This can be required with certain OSes or file systems (e.g.
-///   NFS), but not on Linux with most common file systems like ext4 (which we currently use).
-///
-/// An audit of 8 other databases found that none fsynced the file after a rename:
-/// <https://github.com/neondatabase/neon/pull/9686#discussion_r1837180535>
-///
-/// eBPF probes confirmed that this is sufficient with ext4, XFS, and ZFS, but possibly not Btrfs:
-/// <https://github.com/neondatabase/neon/pull/9686#discussion_r1837926218>
+/// Unlike postgres, it only does fsyncs to 1) file to be renamed to make
+/// contents durable; 2) its directory entry to make rename durable 3) again to
+/// already renamed file, which is not required by standards but postgres does
+/// it, let's stick to that. Postgres additionally fsyncs newpath *before*
+/// rename if it exists to ensure that at least one of the files survives, but
+/// current callers don't need that.
 ///
 /// virtual_file.rs has similar code, but it doesn't use vfs.
 ///
@@ -154,21 +144,35 @@ pub async fn durable_rename(
     old_path: impl AsRef<Utf8Path>,
     new_path: impl AsRef<Utf8Path>,
     do_fsync: bool,
-) -> io::Result<()> {
+) -> io::Result<[Duration; 3]> {
+    async fn maybe_fsync_with_latency(path: &Utf8Path, do_fsync: bool) -> io::Result<Duration> {
+        if !do_fsync {
+            return Ok(Duration::ZERO);
+        }
+        let start = std::time::Instant::now();
+        fsync_async(path).await?;
+        Ok(start.elapsed())
+    }
+
+    let mut latency = [Duration::ZERO; 3];
+
     // first fsync the file
-    fsync_async_opt(old_path.as_ref(), do_fsync).await?;
+    latency[0] = maybe_fsync_with_latency(old_path.as_ref(), do_fsync).await?;
 
     // Time to do the real deal.
     tokio::fs::rename(old_path.as_ref(), new_path.as_ref()).await?;
 
+    // Postgres'ish fsync of renamed file.
+    latency[1] = maybe_fsync_with_latency(new_path.as_ref(), do_fsync).await?;
+
     // Now fsync the parent
     let parent = match new_path.as_ref().parent() {
         Some(p) => p,
         None => Utf8Path::new("./"), // assume current dir if there is no parent
     };
-    fsync_async_opt(parent, do_fsync).await?;
+    latency[2] = maybe_fsync_with_latency(parent, do_fsync).await?;
 
-    Ok(())
+    Ok(latency)
 }
 
 /// Writes a file to the specified `final_path` in a crash safe fasion, using [`std::fs`].

libs/utils/src/lsn.rs

@@ -138,11 +138,6 @@ impl Lsn {
         self.0.checked_sub(other).map(Lsn)
     }
 
-    /// Subtract a number, saturating at numeric bounds instead of overflowing.
-    pub fn saturating_sub<T: Into<u64>>(self, other: T) -> Lsn {
-        Lsn(self.0.saturating_sub(other.into()))
-    }
-
     /// Subtract a number, returning the difference as i128 to avoid overflow.
     pub fn widening_sub<T: Into<u64>>(self, other: T) -> i128 {
         let other: u64 = other.into();

pageserver/compaction/src/helpers.rs

@@ -35,15 +35,6 @@ pub fn overlaps_with<T: Ord>(a: &Range<T>, b: &Range<T>) -> bool {
     !(a.end <= b.start || b.end <= a.start)
 }
 
-/// Whether a fully contains b, example as below
-/// ```plain
-/// |      a       |
-///       |  b  |
-/// ```
-pub fn fully_contains<T: Ord>(a: &Range<T>, b: &Range<T>) -> bool {
-    a.start <= b.start && a.end >= b.end
-}
-
 pub fn union_to_keyspace<K: Ord>(a: &mut CompactionKeySpace<K>, b: CompactionKeySpace<K>) {
     let x = std::mem::take(a);
     let mut all_ranges_iter = [x.into_iter(), b.into_iter()]

pageserver/src/auth.rs

@@ -19,8 +19,7 @@ pub fn check_permission(claims: &Claims, tenant_id: Option<TenantId>) -> Result<
             | Scope::SafekeeperData
             | Scope::GenerationsApi
             | Scope::Infra
-            | Scope::Scrubber
-            | Scope::ControllerPeer,
+            | Scope::Scrubber,
             _,
         ) => Err(AuthError(
             format!(

pageserver/src/bin/pageserver.rs

@@ -154,17 +154,13 @@ fn main() -> anyhow::Result<()> {
             },
         };
 
-        if conf.no_sync {
-            info!("Skipping syncfs on startup");
-        } else {
-            let started = Instant::now();
-            syncfs(dirfd)?;
-            let elapsed = started.elapsed();
-            info!(
-                elapsed_ms = elapsed.as_millis(),
-                "made tenant directory contents durable"
-            );
-        }
+        let started = Instant::now();
+        syncfs(dirfd)?;
+        let elapsed = started.elapsed();
+        info!(
+            elapsed_ms = elapsed.as_millis(),
+            "made tenant directory contents durable"
+        );
     }
 
     // Initialize up failpoints support

pageserver/src/config.rs

@@ -69,7 +69,6 @@ pub struct PageServerConf {
     pub wal_redo_timeout: Duration,
 
     pub superuser: String,
-    pub locale: String,
 
     pub page_cache_size: usize,
     pub max_file_descriptors: usize,
@@ -179,9 +178,6 @@ pub struct PageServerConf {
 
     /// Direct IO settings
     pub virtual_file_io_mode: virtual_file::IoMode,
-
-    /// Optionally disable disk syncs (unsafe!)
-    pub no_sync: bool,
 }
 
 /// Token for authentication to safekeepers
@@ -302,7 +298,6 @@ impl PageServerConf {
             wait_lsn_timeout,
             wal_redo_timeout,
             superuser,
-            locale,
             page_cache_size,
             max_file_descriptors,
             pg_distrib_dir,
@@ -337,7 +332,6 @@ impl PageServerConf {
             concurrent_tenant_size_logical_size_queries,
             virtual_file_io_engine,
             tenant_config,
-            no_sync,
         } = config_toml;
 
         let mut conf = PageServerConf {
@@ -350,7 +344,6 @@ impl PageServerConf {
             wait_lsn_timeout,
             wal_redo_timeout,
             superuser,
-            locale,
             page_cache_size,
             max_file_descriptors,
             http_auth_type,
@@ -416,7 +409,6 @@ impl PageServerConf {
                 .map(crate::l0_flush::L0FlushConfig::from)
                 .unwrap_or_default(),
             virtual_file_io_mode: virtual_file_io_mode.unwrap_or(virtual_file::IoMode::preferred()),
-            no_sync: no_sync.unwrap_or(false),
         };
 
         // ------------------------------------------------------------

pageserver/src/http/routes.rs

@@ -2002,9 +2002,9 @@ async fn timeline_offload_handler(
                 "timeline has attached children".into(),
             ));
         }
-        if let (false, reason) = timeline.can_offload() {
+        if !timeline.can_offload() {
             return Err(ApiError::PreconditionFailed(
-                format!("Timeline::can_offload() check failed: {}", reason) .into(),
+                "Timeline::can_offload() returned false".into(),
             ));
         }
         offload_timeline(&tenant, &timeline)
@@ -2169,21 +2169,6 @@ async fn timeline_detach_ancestor_handler(
         let ctx = RequestContext::new(TaskKind::DetachAncestor, DownloadBehavior::Download);
         let ctx = &ctx;
 
-        // Flush the upload queues of all timelines before detaching ancestor. We do the same thing again
-        // during shutdown. This early upload ensures the pageserver does not need to upload too many
-        // things and creates downtime during timeline reloads.
-        for timeline in tenant.list_timelines() {
-            timeline
-                .remote_client
-                .wait_completion()
-                .await
-                .map_err(|e| {
-                    ApiError::PreconditionFailed(format!("cannot drain upload queue: {e}").into())
-                })?;
-        }
-
-        tracing::info!("all timeline upload queues are drained");
-
         let timeline = tenant.get_timeline(timeline_id, true)?;
 
         let progress = timeline

pageserver/src/pgdatadir_mapping.rs

@@ -45,7 +45,7 @@ use wal_decoder::serialized_batch::SerializedValueBatch;
 pub const MAX_AUX_FILE_DELTAS: usize = 1024;
 
 /// Max number of aux-file-related delta layers. The compaction will create a new image layer once this threshold is reached.
-pub const MAX_AUX_FILE_V2_DELTAS: usize = 16;
+pub const MAX_AUX_FILE_V2_DELTAS: usize = 64;
 
 #[derive(Debug)]
 pub enum LsnForTimestamp {

pageserver/src/tenant.rs

@@ -2493,8 +2493,7 @@ impl Tenant {
             timelines_to_compact_or_offload = timelines
                 .iter()
                 .filter_map(|(timeline_id, timeline)| {
-                    let (is_active, (can_offload, _)) =
-                        (timeline.is_active(), timeline.can_offload());
+                    let (is_active, can_offload) = (timeline.is_active(), timeline.can_offload());
                     let has_no_unoffloaded_children = {
                         !timelines
                             .iter()
@@ -4780,12 +4779,10 @@ async fn run_initdb(
 
     let _permit = INIT_DB_SEMAPHORE.acquire().await;
 
-    let mut initdb_command = tokio::process::Command::new(&initdb_bin_path);
-    initdb_command
+    let initdb_command = tokio::process::Command::new(&initdb_bin_path)
         .args(["--pgdata", initdb_target_dir.as_ref()])
         .args(["--username", &conf.superuser])
         .args(["--encoding", "utf8"])
-        .args(["--locale", &conf.locale])
         .arg("--no-instructions")
         .arg("--no-sync")
         .env_clear()
@@ -4795,27 +4792,15 @@ async fn run_initdb(
         // stdout invocation produces the same output every time, we don't need it
         .stdout(std::process::Stdio::null())
         // we would be interested in the stderr output, if there was any
-        .stderr(std::process::Stdio::piped());
-
-    // Before version 14, only the libc provide was available.
-    if pg_version > 14 {
-        // Version 17 brought with it a builtin locale provider which only provides
-        // C and C.UTF-8. While being safer for collation purposes since it is
-        // guaranteed to be consistent throughout a major release, it is also more
-        // performant.
-        let locale_provider = if pg_version >= 17 { "builtin" } else { "libc" };
-
-        initdb_command.args(["--locale-provider", locale_provider]);
-    }
-
-    let initdb_proc = initdb_command.spawn()?;
+        .stderr(std::process::Stdio::piped())
+        .spawn()?;
 
     // Ideally we'd select here with the cancellation token, but the problem is that
     // we can't safely terminate initdb: it launches processes of its own, and killing
     // initdb doesn't kill them. After we return from this function, we want the target
     // directory to be able to be cleaned up.
     // See https://github.com/neondatabase/neon/issues/6385
-    let initdb_output = initdb_proc.wait_with_output().await?;
+    let initdb_output = initdb_command.wait_with_output().await?;
     if !initdb_output.status.success() {
         return Err(InitdbError::Failed(
             initdb_output.status,
@@ -7757,13 +7742,13 @@ mod tests {
             (
                 get_key(3),
                 Lsn(0x20),
-                Value::WalRecord(NeonWalRecord::wal_clear("c")),
+                Value::WalRecord(NeonWalRecord::wal_clear()),
             ),
             (get_key(4), Lsn(0x10), Value::Image("0x10".into())),
             (
                 get_key(4),
                 Lsn(0x20),
-                Value::WalRecord(NeonWalRecord::wal_init("i")),
+                Value::WalRecord(NeonWalRecord::wal_init()),
             ),
         ];
         let image1 = vec![(get_key(1), "0x10".into())];
@@ -7912,30 +7897,8 @@ mod tests {
 
     #[cfg(feature = "testing")]
     #[tokio::test]
-    async fn test_simple_bottom_most_compaction_deltas_1() -> anyhow::Result<()> {
-        test_simple_bottom_most_compaction_deltas_helper(
-            "test_simple_bottom_most_compaction_deltas_1",
-            false,
-        )
-        .await
-    }
-
-    #[cfg(feature = "testing")]
-    #[tokio::test]
-    async fn test_simple_bottom_most_compaction_deltas_2() -> anyhow::Result<()> {
-        test_simple_bottom_most_compaction_deltas_helper(
-            "test_simple_bottom_most_compaction_deltas_2",
-            true,
-        )
-        .await
-    }
-
-    #[cfg(feature = "testing")]
-    async fn test_simple_bottom_most_compaction_deltas_helper(
-        test_name: &'static str,
-        use_delta_bottom_layer: bool,
-    ) -> anyhow::Result<()> {
-        let harness = TenantHarness::create(test_name).await?;
+    async fn test_simple_bottom_most_compaction_deltas() -> anyhow::Result<()> {
+        let harness = TenantHarness::create("test_simple_bottom_most_compaction_deltas").await?;
         let (tenant, ctx) = harness.load().await;
 
         fn get_key(id: u32) -> Key {
@@ -7966,16 +7929,6 @@ mod tests {
         let img_layer = (0..10)
             .map(|id| (get_key(id), Bytes::from(format!("value {id}@0x10"))))
             .collect_vec();
-        // or, delta layer at 0x10 if `use_delta_bottom_layer` is true
-        let delta4 = (0..10)
-            .map(|id| {
-                (
-                    get_key(id),
-                    Lsn(0x08),
-                    Value::WalRecord(NeonWalRecord::wal_init(format!("value {id}@0x10"))),
-                )
-            })
-            .collect_vec();
 
         let delta1 = vec![
             (
@@ -8029,61 +7982,21 @@ mod tests {
             ),
         ];
 
-        let tline = if use_delta_bottom_layer {
-            tenant
-                .create_test_timeline_with_layers(
-                    TIMELINE_ID,
-                    Lsn(0x08),
-                    DEFAULT_PG_VERSION,
-                    &ctx,
-                    vec![
-                        DeltaLayerTestDesc::new_with_inferred_key_range(
-                            Lsn(0x08)..Lsn(0x10),
-                            delta4,
-                        ),
-                        DeltaLayerTestDesc::new_with_inferred_key_range(
-                            Lsn(0x20)..Lsn(0x48),
-                            delta1,
-                        ),
-                        DeltaLayerTestDesc::new_with_inferred_key_range(
-                            Lsn(0x20)..Lsn(0x48),
-                            delta2,
-                        ),
-                        DeltaLayerTestDesc::new_with_inferred_key_range(
-                            Lsn(0x48)..Lsn(0x50),
-                            delta3,
-                        ),
-                    ], // delta layers
-                    vec![], // image layers
-                    Lsn(0x50),
-                )
-                .await?
-        } else {
-            tenant
-                .create_test_timeline_with_layers(
-                    TIMELINE_ID,
-                    Lsn(0x10),
-                    DEFAULT_PG_VERSION,
-                    &ctx,
-                    vec![
-                        DeltaLayerTestDesc::new_with_inferred_key_range(
-                            Lsn(0x10)..Lsn(0x48),
-                            delta1,
-                        ),
-                        DeltaLayerTestDesc::new_with_inferred_key_range(
-                            Lsn(0x10)..Lsn(0x48),
-                            delta2,
-                        ),
-                        DeltaLayerTestDesc::new_with_inferred_key_range(
-                            Lsn(0x48)..Lsn(0x50),
-                            delta3,
-                        ),
-                    ], // delta layers
-                    vec![(Lsn(0x10), img_layer)], // image layers
-                    Lsn(0x50),
-                )
-                .await?
-        };
+        let tline = tenant
+            .create_test_timeline_with_layers(
+                TIMELINE_ID,
+                Lsn(0x10),
+                DEFAULT_PG_VERSION,
+                &ctx,
+                vec![
+                    DeltaLayerTestDesc::new_with_inferred_key_range(Lsn(0x10)..Lsn(0x48), delta1),
+                    DeltaLayerTestDesc::new_with_inferred_key_range(Lsn(0x10)..Lsn(0x48), delta2),
+                    DeltaLayerTestDesc::new_with_inferred_key_range(Lsn(0x48)..Lsn(0x50), delta3),
+                ], // delta layers
+                vec![(Lsn(0x10), img_layer)], // image layers
+                Lsn(0x50),
+            )
+            .await?;
         {
             // Update GC info
             let mut guard = tline.gc_info.write().unwrap();
@@ -8193,7 +8106,7 @@ mod tests {
             (
                 key,
                 Lsn(0x10),
-                Value::WalRecord(NeonWalRecord::wal_init("0x10")),
+                Value::Image(Bytes::copy_from_slice(b"0x10")),
             ),
             (
                 key,
@@ -8255,7 +8168,7 @@ mod tests {
                     Lsn(0x20),
                     KeyLogAtLsn(vec![(
                         Lsn(0x20),
-                        Value::Image(Bytes::from_static(b"0x10;0x20")),
+                        Value::Image(Bytes::copy_from_slice(b"0x10;0x20")),
                     )]),
                 ),
                 (
@@ -9237,7 +9150,7 @@ mod tests {
 
             let will_init = will_init_keys.contains(&i);
             if will_init {
-                delta_layer_spec.push((key, lsn, Value::WalRecord(NeonWalRecord::wal_init(""))));
+                delta_layer_spec.push((key, lsn, Value::WalRecord(NeonWalRecord::wal_init())));
 
                 expected_key_values.insert(key, "".to_string());
             } else {
@@ -9295,23 +9208,6 @@ mod tests {
         Ok(())
     }
 
-    fn sort_layer_key(k1: &PersistentLayerKey, k2: &PersistentLayerKey) -> std::cmp::Ordering {
-        (
-            k1.is_delta,
-            k1.key_range.start,
-            k1.key_range.end,
-            k1.lsn_range.start,
-            k1.lsn_range.end,
-        )
-            .cmp(&(
-                k2.is_delta,
-                k2.key_range.start,
-                k2.key_range.end,
-                k2.lsn_range.start,
-                k2.lsn_range.end,
-            ))
-    }
-
     async fn inspect_and_sort(
         tline: &Arc<Timeline>,
         filter: Option<std::ops::Range<Key>>,
@@ -9320,30 +9216,25 @@ mod tests {
         if let Some(filter) = filter {
             all_layers.retain(|layer| overlaps_with(&layer.key_range, &filter));
         }
-        all_layers.sort_by(sort_layer_key);
+        all_layers.sort_by(|k1, k2| {
+            (
+                k1.is_delta,
+                k1.key_range.start,
+                k1.key_range.end,
+                k1.lsn_range.start,
+                k1.lsn_range.end,
+            )
+                .cmp(&(
+                    k2.is_delta,
+                    k2.key_range.start,
+                    k2.key_range.end,
+                    k2.lsn_range.start,
+                    k2.lsn_range.end,
+                ))
+        });
         all_layers
     }
 
-    #[cfg(feature = "testing")]
-    fn check_layer_map_key_eq(
-        mut left: Vec<PersistentLayerKey>,
-        mut right: Vec<PersistentLayerKey>,
-    ) {
-        left.sort_by(sort_layer_key);
-        right.sort_by(sort_layer_key);
-        if left != right {
-            eprintln!("---LEFT---");
-            for left in left.iter() {
-                eprintln!("{}", left);
-            }
-            eprintln!("---RIGHT---");
-            for right in right.iter() {
-                eprintln!("{}", right);
-            }
-            assert_eq!(left, right);
-        }
-    }
-
     #[cfg(feature = "testing")]
     #[tokio::test]
     async fn test_simple_partial_bottom_most_compaction() -> anyhow::Result<()> {
@@ -9436,206 +9327,127 @@ mod tests {
 
         let cancel = CancellationToken::new();
 
-        // Do a partial compaction on key range 0..2
+        // Do a partial compaction on key range 0..4, we should generate a image layer; no other layers
+        // can be removed because they might be used for other key ranges.
         tline
-            .partial_compact_with_gc(get_key(0)..get_key(2), &cancel, EnumSet::new(), &ctx)
+            .partial_compact_with_gc(Some(get_key(0)..get_key(4)), &cancel, EnumSet::new(), &ctx)
             .await
             .unwrap();
         let all_layers = inspect_and_sort(&tline, Some(get_key(0)..get_key(10))).await;
-        check_layer_map_key_eq(
+        assert_eq!(
             all_layers,
             vec![
-                // newly-generated image layer for the partial compaction range 0-2
                 PersistentLayerKey {
-                    key_range: get_key(0)..get_key(2),
+                    key_range: get_key(0)..get_key(4),
                     lsn_range: Lsn(0x20)..Lsn(0x21),
-                    is_delta: false,
+                    is_delta: false
                 },
                 PersistentLayerKey {
                     key_range: get_key(0)..get_key(10),
                     lsn_range: Lsn(0x10)..Lsn(0x11),
-                    is_delta: false,
+                    is_delta: false
                 },
-                // delta1 is split and the second part is rewritten
                 PersistentLayerKey {
-                    key_range: get_key(2)..get_key(4),
+                    key_range: get_key(1)..get_key(4),
                     lsn_range: Lsn(0x20)..Lsn(0x48),
-                    is_delta: true,
+                    is_delta: true
                 },
                 PersistentLayerKey {
                     key_range: get_key(5)..get_key(7),
                     lsn_range: Lsn(0x20)..Lsn(0x48),
-                    is_delta: true,
+                    is_delta: true
                 },
                 PersistentLayerKey {
                     key_range: get_key(8)..get_key(10),
                     lsn_range: Lsn(0x48)..Lsn(0x50),
-                    is_delta: true,
-                },
-            ],
+                    is_delta: true
+                }
+            ]
         );
 
-        // Do a partial compaction on key range 2..4
+        // Do a partial compaction on key range 4..10
         tline
-            .partial_compact_with_gc(get_key(2)..get_key(4), &cancel, EnumSet::new(), &ctx)
+            .partial_compact_with_gc(Some(get_key(4)..get_key(10)), &cancel, EnumSet::new(), &ctx)
             .await
             .unwrap();
         let all_layers = inspect_and_sort(&tline, Some(get_key(0)..get_key(10))).await;
-        check_layer_map_key_eq(
+        assert_eq!(
             all_layers,
             vec![
                 PersistentLayerKey {
-                    key_range: get_key(0)..get_key(2),
+                    key_range: get_key(0)..get_key(4),
                     lsn_range: Lsn(0x20)..Lsn(0x21),
-                    is_delta: false,
+                    is_delta: false
                 },
                 PersistentLayerKey {
+                    // if (in the future) GC kicks in, this layer will be removed
                     key_range: get_key(0)..get_key(10),
                     lsn_range: Lsn(0x10)..Lsn(0x11),
-                    is_delta: false,
+                    is_delta: false
                 },
-                // image layer generated for the compaction range 2-4
                 PersistentLayerKey {
-                    key_range: get_key(2)..get_key(4),
+                    key_range: get_key(4)..get_key(10),
                     lsn_range: Lsn(0x20)..Lsn(0x21),
-                    is_delta: false,
+                    is_delta: false
                 },
-                // we have key2/key3 above the retain_lsn, so we still need this delta layer
                 PersistentLayerKey {
-                    key_range: get_key(2)..get_key(4),
+                    key_range: get_key(1)..get_key(4),
                     lsn_range: Lsn(0x20)..Lsn(0x48),
-                    is_delta: true,
+                    is_delta: true
                 },
                 PersistentLayerKey {
                     key_range: get_key(5)..get_key(7),
                     lsn_range: Lsn(0x20)..Lsn(0x48),
-                    is_delta: true,
+                    is_delta: true
                 },
                 PersistentLayerKey {
                     key_range: get_key(8)..get_key(10),
                     lsn_range: Lsn(0x48)..Lsn(0x50),
-                    is_delta: true,
-                },
-            ],
-        );
-
-        // Do a partial compaction on key range 4..9
-        tline
-            .partial_compact_with_gc(get_key(4)..get_key(9), &cancel, EnumSet::new(), &ctx)
-            .await
-            .unwrap();
-        let all_layers = inspect_and_sort(&tline, Some(get_key(0)..get_key(10))).await;
-        check_layer_map_key_eq(
-            all_layers,
-            vec![
-                PersistentLayerKey {
-                    key_range: get_key(0)..get_key(2),
-                    lsn_range: Lsn(0x20)..Lsn(0x21),
-                    is_delta: false,
-                },
-                PersistentLayerKey {
-                    key_range: get_key(0)..get_key(10),
-                    lsn_range: Lsn(0x10)..Lsn(0x11),
-                    is_delta: false,
-                },
-                PersistentLayerKey {
-                    key_range: get_key(2)..get_key(4),
-                    lsn_range: Lsn(0x20)..Lsn(0x21),
-                    is_delta: false,
-                },
-                PersistentLayerKey {
-                    key_range: get_key(2)..get_key(4),
-                    lsn_range: Lsn(0x20)..Lsn(0x48),
-                    is_delta: true,
-                },
-                // image layer generated for this compaction range
-                PersistentLayerKey {
-                    key_range: get_key(4)..get_key(9),
-                    lsn_range: Lsn(0x20)..Lsn(0x21),
-                    is_delta: false,
-                },
-                PersistentLayerKey {
-                    key_range: get_key(8)..get_key(10),
-                    lsn_range: Lsn(0x48)..Lsn(0x50),
-                    is_delta: true,
-                },
-            ],
-        );
-
-        // Do a partial compaction on key range 9..10
-        tline
-            .partial_compact_with_gc(get_key(9)..get_key(10), &cancel, EnumSet::new(), &ctx)
-            .await
-            .unwrap();
-        let all_layers = inspect_and_sort(&tline, Some(get_key(0)..get_key(10))).await;
-        check_layer_map_key_eq(
-            all_layers,
-            vec![
-                PersistentLayerKey {
-                    key_range: get_key(0)..get_key(2),
-                    lsn_range: Lsn(0x20)..Lsn(0x21),
-                    is_delta: false,
-                },
-                PersistentLayerKey {
-                    key_range: get_key(0)..get_key(10),
-                    lsn_range: Lsn(0x10)..Lsn(0x11),
-                    is_delta: false,
-                },
-                PersistentLayerKey {
-                    key_range: get_key(2)..get_key(4),
-                    lsn_range: Lsn(0x20)..Lsn(0x21),
-                    is_delta: false,
-                },
-                PersistentLayerKey {
-                    key_range: get_key(2)..get_key(4),
-                    lsn_range: Lsn(0x20)..Lsn(0x48),
-                    is_delta: true,
-                },
-                PersistentLayerKey {
-                    key_range: get_key(4)..get_key(9),
-                    lsn_range: Lsn(0x20)..Lsn(0x21),
-                    is_delta: false,
-                },
-                // image layer generated for the compaction range
-                PersistentLayerKey {
-                    key_range: get_key(9)..get_key(10),
-                    lsn_range: Lsn(0x20)..Lsn(0x21),
-                    is_delta: false,
-                },
-                PersistentLayerKey {
-                    key_range: get_key(8)..get_key(10),
-                    lsn_range: Lsn(0x48)..Lsn(0x50),
-                    is_delta: true,
-                },
-            ],
+                    is_delta: true
+                }
+            ]
         );
 
         // Do a partial compaction on key range 0..10, all image layers below LSN 20 can be replaced with new ones.
         tline
-            .partial_compact_with_gc(get_key(0)..get_key(10), &cancel, EnumSet::new(), &ctx)
+            .partial_compact_with_gc(Some(get_key(0)..get_key(10)), &cancel, EnumSet::new(), &ctx)
             .await
             .unwrap();
         let all_layers = inspect_and_sort(&tline, Some(get_key(0)..get_key(10))).await;
-        check_layer_map_key_eq(
+        assert_eq!(
             all_layers,
             vec![
-                // aha, we removed all unnecessary image/delta layers and got a very clean layer map!
+                PersistentLayerKey {
+                    key_range: get_key(0)..get_key(4),
+                    lsn_range: Lsn(0x20)..Lsn(0x21),
+                    is_delta: false
+                },
                 PersistentLayerKey {
                     key_range: get_key(0)..get_key(10),
                     lsn_range: Lsn(0x20)..Lsn(0x21),
-                    is_delta: false,
+                    is_delta: false
                 },
                 PersistentLayerKey {
-                    key_range: get_key(2)..get_key(4),
+                    key_range: get_key(4)..get_key(10),
+                    lsn_range: Lsn(0x20)..Lsn(0x21),
+                    is_delta: false
+                },
+                PersistentLayerKey {
+                    key_range: get_key(1)..get_key(4),
                     lsn_range: Lsn(0x20)..Lsn(0x48),
-                    is_delta: true,
+                    is_delta: true
+                },
+                PersistentLayerKey {
+                    key_range: get_key(5)..get_key(7),
+                    lsn_range: Lsn(0x20)..Lsn(0x48),
+                    is_delta: true
                 },
                 PersistentLayerKey {
                     key_range: get_key(8)..get_key(10),
                     lsn_range: Lsn(0x48)..Lsn(0x50),
-                    is_delta: true,
-                },
-            ],
+                    is_delta: true
+                }
+            ]
         );
 
         Ok(())

pageserver/src/tenant/mgr.rs

@@ -1959,7 +1959,7 @@ impl TenantManager {
             attempt.before_reset_tenant();
 
             let (_guard, progress) = utils::completion::channel();
-            match tenant.shutdown(progress, ShutdownMode::Flush).await {
+            match tenant.shutdown(progress, ShutdownMode::Hard).await {
                 Ok(()) => {
                     slot_guard.drop_old_value().expect("it was just shutdown");
                 }

pageserver/src/tenant/remote_timeline_client.rs

@@ -2201,18 +2201,6 @@ impl RemoteTimelineClient {
         inner.initialized_mut()?;
         Ok(UploadQueueAccessor { inner })
     }
-
-    pub(crate) fn no_pending_work(&self) -> bool {
-        let inner = self.upload_queue.lock().unwrap();
-        match &*inner {
-            UploadQueue::Uninitialized
-            | UploadQueue::Stopped(UploadQueueStopped::Uninitialized) => true,
-            UploadQueue::Stopped(UploadQueueStopped::Deletable(x)) => {
-                x.upload_queue_for_deletion.no_pending_work()
-            }
-            UploadQueue::Initialized(x) => x.no_pending_work(),
-        }
-    }
 }
 
 pub(crate) struct UploadQueueAccessor<'a> {

pageserver/src/tenant/storage_layer/delta_layer.rs

@@ -653,10 +653,6 @@ impl DeltaLayerWriter {
         })
     }
 
-    pub fn is_empty(&self) -> bool {
-        self.inner.as_ref().unwrap().num_keys == 0
-    }
-
     ///
     /// Append a key-value pair to the file.
     ///

pageserver/src/tenant/storage_layer/filter_iterator.rs

@@ -1,4 +1,4 @@
-use std::{ops::Range, sync::Arc};
+use std::ops::Range;
 
 use anyhow::bail;
 use pageserver_api::{
@@ -9,10 +9,7 @@ use utils::lsn::Lsn;
 
 use pageserver_api::value::Value;
 
-use super::{
-    merge_iterator::{MergeIterator, MergeIteratorItem},
-    PersistentLayerKey,
-};
+use super::merge_iterator::MergeIterator;
 
 /// A filter iterator over merge iterators (and can be easily extended to other types of iterators).
 ///
@@ -51,10 +48,10 @@ impl<'a> FilterIterator<'a> {
         })
     }
 
-    async fn next_inner<R: MergeIteratorItem>(&mut self) -> anyhow::Result<Option<R>> {
-        while let Some(item) = self.inner.next_inner::<R>().await? {
+    pub async fn next(&mut self) -> anyhow::Result<Option<(Key, Lsn, Value)>> {
+        while let Some(item) = self.inner.next().await? {
             while self.current_filter_idx < self.retain_key_filters.len()
-                && item.key_lsn_value().0 >= self.retain_key_filters[self.current_filter_idx].end
+                && item.0 >= self.retain_key_filters[self.current_filter_idx].end
             {
                 // [filter region]    [filter region]     [filter region]
                 //                                     ^ item
@@ -71,7 +68,7 @@ impl<'a> FilterIterator<'a> {
                 //                                                 ^ current filter (nothing)
                 return Ok(None);
             }
-            if self.retain_key_filters[self.current_filter_idx].contains(&item.key_lsn_value().0) {
+            if self.retain_key_filters[self.current_filter_idx].contains(&item.0) {
                 // [filter region]    [filter region]     [filter region]
                 //                                              ^ item
                 //                                        ^ current filter
@@ -84,16 +81,6 @@ impl<'a> FilterIterator<'a> {
         }
         Ok(None)
     }
-
-    pub async fn next(&mut self) -> anyhow::Result<Option<(Key, Lsn, Value)>> {
-        self.next_inner().await
-    }
-
-    pub async fn next_with_trace(
-        &mut self,
-    ) -> anyhow::Result<Option<((Key, Lsn, Value), Arc<PersistentLayerKey>)>> {
-        self.next_inner().await
-    }
 }
 
 #[cfg(test)]

pageserver/src/tenant/storage_layer/inmemory_layer.rs

@@ -67,8 +67,6 @@ pub struct InMemoryLayer {
     /// The above fields never change, except for `end_lsn`, which is only set once.
     /// All other changing parts are in `inner`, and protected by a mutex.
     inner: RwLock<InMemoryLayerInner>,
-
-    estimated_in_mem_size: AtomicU64,
 }
 
 impl std::fmt::Debug for InMemoryLayer {
@@ -545,10 +543,6 @@ impl InMemoryLayer {
         Ok(inner.file.len())
     }
 
-    pub fn estimated_in_mem_size(&self) -> u64 {
-        self.estimated_in_mem_size.load(AtomicOrdering::Relaxed)
-    }
-
     /// Create a new, empty, in-memory layer
     pub async fn create(
         conf: &'static PageServerConf,
@@ -578,7 +572,6 @@ impl InMemoryLayer {
                 file,
                 resource_units: GlobalResourceUnits::new(),
             }),
-            estimated_in_mem_size: AtomicU64::new(0),
         })
     }
 
@@ -649,12 +642,6 @@ impl InMemoryLayer {
                 // because this case is unexpected, and we would like tests to fail if this happens.
                 warn!("Key {} at {} written twice at same LSN", key, lsn);
             }
-            self.estimated_in_mem_size.fetch_add(
-                (std::mem::size_of::<CompactKey>()
-                    + std::mem::size_of::<Lsn>()
-                    + std::mem::size_of::<IndexEntry>()) as u64,
-                AtomicOrdering::Relaxed,
-            );
         }
 
         inner.resource_units.maybe_publish_size(new_size);

pageserver/src/tenant/storage_layer/merge_iterator.rs

@@ -1,7 +1,6 @@
 use std::{
     cmp::Ordering,
     collections::{binary_heap, BinaryHeap},
-    sync::Arc,
 };
 
 use anyhow::bail;
@@ -14,11 +13,10 @@ use pageserver_api::value::Value;
 use super::{
     delta_layer::{DeltaLayerInner, DeltaLayerIterator},
     image_layer::{ImageLayerInner, ImageLayerIterator},
-    PersistentLayerDesc, PersistentLayerKey,
 };
 
 #[derive(Clone, Copy)]
-pub(crate) enum LayerRef<'a> {
+enum LayerRef<'a> {
     Image(&'a ImageLayerInner),
     Delta(&'a DeltaLayerInner),
 }
@@ -64,20 +62,18 @@ impl LayerIterRef<'_> {
 /// 1. Unified iterator for image and delta layers.
 /// 2. `Ord` for use in [`MergeIterator::heap`] (for the k-merge).
 /// 3. Lazy creation of the real delta/image iterator.
-pub(crate) enum IteratorWrapper<'a> {
+enum IteratorWrapper<'a> {
     NotLoaded {
         ctx: &'a RequestContext,
         first_key_lower_bound: (Key, Lsn),
         layer: LayerRef<'a>,
-        source_desc: Arc<PersistentLayerKey>,
     },
     Loaded {
         iter: PeekableLayerIterRef<'a>,
-        source_desc: Arc<PersistentLayerKey>,
     },
 }
 
-pub(crate) struct PeekableLayerIterRef<'a> {
+struct PeekableLayerIterRef<'a> {
     iter: LayerIterRef<'a>,
     peeked: Option<(Key, Lsn, Value)>, // None == end
 }
@@ -155,12 +151,6 @@ impl<'a> IteratorWrapper<'a> {
             layer: LayerRef::Image(image_layer),
             first_key_lower_bound: (image_layer.key_range().start, image_layer.lsn()),
             ctx,
-            source_desc: PersistentLayerKey {
-                key_range: image_layer.key_range().clone(),
-                lsn_range: PersistentLayerDesc::image_layer_lsn_range(image_layer.lsn()),
-                is_delta: false,
-            }
-            .into(),
         }
     }
 
@@ -172,18 +162,12 @@ impl<'a> IteratorWrapper<'a> {
             layer: LayerRef::Delta(delta_layer),
             first_key_lower_bound: (delta_layer.key_range().start, delta_layer.lsn_range().start),
             ctx,
-            source_desc: PersistentLayerKey {
-                key_range: delta_layer.key_range().clone(),
-                lsn_range: delta_layer.lsn_range().clone(),
-                is_delta: true,
-            }
-            .into(),
         }
     }
 
     fn peek_next_key_lsn_value(&self) -> Option<(&Key, Lsn, Option<&Value>)> {
         match self {
-            Self::Loaded { iter, .. } => iter
+            Self::Loaded { iter } => iter
                 .peek()
                 .as_ref()
                 .map(|(key, lsn, val)| (key, *lsn, Some(val))),
@@ -207,7 +191,6 @@ impl<'a> IteratorWrapper<'a> {
             ctx,
             first_key_lower_bound,
             layer,
-            source_desc,
         } = self
         else {
             unreachable!()
@@ -223,10 +206,7 @@ impl<'a> IteratorWrapper<'a> {
                 );
             }
         }
-        *self = Self::Loaded {
-            iter,
-            source_desc: source_desc.clone(),
-        };
+        *self = Self::Loaded { iter };
         Ok(())
     }
 
@@ -240,19 +220,11 @@ impl<'a> IteratorWrapper<'a> {
     /// The public interfaces to use are [`crate::tenant::storage_layer::delta_layer::DeltaLayerIterator`] and
     /// [`crate::tenant::storage_layer::image_layer::ImageLayerIterator`].
     async fn next(&mut self) -> anyhow::Result<Option<(Key, Lsn, Value)>> {
-        let Self::Loaded { iter, .. } = self else {
+        let Self::Loaded { iter } = self else {
             panic!("must load the iterator before using")
         };
         iter.next().await
     }
-
-    /// Get the persistent layer key corresponding to this iterator
-    fn trace_source(&self) -> Arc<PersistentLayerKey> {
-        match self {
-            Self::Loaded { source_desc, .. } => source_desc.clone(),
-            Self::NotLoaded { source_desc, .. } => source_desc.clone(),
-        }
-    }
 }
 
 /// A merge iterator over delta/image layer iterators.
@@ -270,32 +242,6 @@ pub struct MergeIterator<'a> {
     heap: BinaryHeap<IteratorWrapper<'a>>,
 }
 
-pub(crate) trait MergeIteratorItem {
-    fn new(item: (Key, Lsn, Value), iterator: &IteratorWrapper<'_>) -> Self;
-
-    fn key_lsn_value(&self) -> &(Key, Lsn, Value);
-}
-
-impl MergeIteratorItem for (Key, Lsn, Value) {
-    fn new(item: (Key, Lsn, Value), _: &IteratorWrapper<'_>) -> Self {
-        item
-    }
-
-    fn key_lsn_value(&self) -> &(Key, Lsn, Value) {
-        self
-    }
-}
-
-impl MergeIteratorItem for ((Key, Lsn, Value), Arc<PersistentLayerKey>) {
-    fn new(item: (Key, Lsn, Value), iter: &IteratorWrapper<'_>) -> Self {
-        (item, iter.trace_source().clone())
-    }
-
-    fn key_lsn_value(&self) -> &(Key, Lsn, Value) {
-        &self.0
-    }
-}
-
 impl<'a> MergeIterator<'a> {
     pub fn create(
         deltas: &[&'a DeltaLayerInner],
@@ -314,7 +260,7 @@ impl<'a> MergeIterator<'a> {
         }
     }
 
-    pub(crate) async fn next_inner<R: MergeIteratorItem>(&mut self) -> anyhow::Result<Option<R>> {
+    pub async fn next(&mut self) -> anyhow::Result<Option<(Key, Lsn, Value)>> {
         while let Some(mut iter) = self.heap.peek_mut() {
             if !iter.is_loaded() {
                 // Once we load the iterator, we can know the real first key-value pair in the iterator.
@@ -329,22 +275,10 @@ impl<'a> MergeIterator<'a> {
                 binary_heap::PeekMut::pop(iter);
                 continue;
             };
-            return Ok(Some(R::new(item, &iter)));
+            return Ok(Some(item));
         }
         Ok(None)
     }
-
-    /// Get the next key-value pair from the iterator.
-    pub async fn next(&mut self) -> anyhow::Result<Option<(Key, Lsn, Value)>> {
-        self.next_inner().await
-    }
-
-    /// Get the next key-value pair from the iterator, and trace where the key comes from.
-    pub async fn next_with_trace(
-        &mut self,
-    ) -> anyhow::Result<Option<((Key, Lsn, Value), Arc<PersistentLayerKey>)>> {
-        self.next_inner().await
-    }
 }
 
 #[cfg(test)]
@@ -562,7 +496,7 @@ mod tests {
             (
                 get_key(0),
                 Lsn(0x10),
-                Value::WalRecord(NeonWalRecord::wal_init("")),
+                Value::WalRecord(NeonWalRecord::wal_init()),
             ),
             (
                 get_key(0),
@@ -572,7 +506,7 @@ mod tests {
             (
                 get_key(5),
                 Lsn(0x10),
-                Value::WalRecord(NeonWalRecord::wal_init("")),
+                Value::WalRecord(NeonWalRecord::wal_init()),
             ),
             (
                 get_key(5),

pageserver/src/tenant/timeline.rs

@@ -23,7 +23,6 @@ use handle::ShardTimelineId;
 use offload::OffloadError;
 use once_cell::sync::Lazy;
 use pageserver_api::{
-    config::tenant_conf_defaults::DEFAULT_COMPACTION_THRESHOLD,
     key::{
         KEY_SIZE, METADATA_KEY_BEGIN_PREFIX, METADATA_KEY_END_PREFIX, NON_INHERITED_RANGE,
         NON_INHERITED_SPARSE_RANGE,
@@ -853,10 +852,6 @@ pub(crate) enum ShutdownMode {
     /// While we are flushing, we continue to accept read I/O for LSNs ingested before
     /// the call to [`Timeline::shutdown`].
     FreezeAndFlush,
-    /// Only flush the layers to the remote storage without freezing any open layers. This is the
-    /// mode used by ancestor detach and any other operations that reloads a tenant but not increasing
-    /// the generation number.
-    Flush,
     /// Shut down immediately, without waiting for any open layers to flush.
     Hard,
 }
@@ -1570,16 +1565,12 @@ impl Timeline {
     ///
     /// This is neccessary but not sufficient for offloading of the timeline as it might have
     /// child timelines that are not offloaded yet.
-    pub(crate) fn can_offload(&self) -> (bool, &'static str) {
+    pub(crate) fn can_offload(&self) -> bool {
         if self.remote_client.is_archived() != Some(true) {
-            return (false, "the timeline is not archived");
-        }
-        if !self.remote_client.no_pending_work() {
-            // if the remote client is still processing some work, we can't offload
-            return (false, "the upload queue is not drained yet");
+            return false;
         }
 
-        (true, "ok")
+        true
     }
 
     /// Outermost timeline compaction operation; downloads needed layers. Returns whether we have pending
@@ -1687,6 +1678,11 @@ impl Timeline {
     pub(crate) async fn shutdown(&self, mode: ShutdownMode) {
         debug_assert_current_span_has_tenant_and_timeline_id();
 
+        let try_freeze_and_flush = match mode {
+            ShutdownMode::FreezeAndFlush => true,
+            ShutdownMode::Hard => false,
+        };
+
         // Regardless of whether we're going to try_freeze_and_flush
         // or not, stop ingesting any more data. Walreceiver only provides
         // cancellation but no "wait until gone", because it uses the Timeline::gate.
@@ -1708,7 +1704,7 @@ impl Timeline {
         // ... and inform any waiters for newer LSNs that there won't be any.
         self.last_record_lsn.shutdown();
 
-        if let ShutdownMode::FreezeAndFlush = mode {
+        if try_freeze_and_flush {
             if let Some((open, frozen)) = self
                 .layers
                 .read()
@@ -1750,20 +1746,6 @@ impl Timeline {
                     warn!("failed to freeze and flush: {e:#}");
                 }
             }
-
-            // `self.remote_client.shutdown().await` above should have already flushed everything from the queue, but
-            // we also do a final check here to ensure that the queue is empty.
-            if !self.remote_client.no_pending_work() {
-                warn!("still have pending work in remote upload queue, but continuing shutting down anyways");
-            }
-        }
-
-        if let ShutdownMode::Flush = mode {
-            // drain the upload queue
-            self.remote_client.shutdown().await;
-            if !self.remote_client.no_pending_work() {
-                warn!("still have pending work in remote upload queue, but continuing shutting down anyways");
-            }
         }
 
         // Signal any subscribers to our cancellation token to drop out
@@ -3506,37 +3488,18 @@ impl Timeline {
 
                 let timer = self.metrics.flush_time_histo.start_timer();
 
-                let num_frozen_layers;
-                let frozen_layer_total_size;
                 let layer_to_flush = {
                     let guard = self.layers.read().await;
                     let Ok(lm) = guard.layer_map() else {
                         info!("dropping out of flush loop for timeline shutdown");
                         return;
                     };
-                    num_frozen_layers = lm.frozen_layers.len();
-                    frozen_layer_total_size = lm
-                        .frozen_layers
-                        .iter()
-                        .map(|l| l.estimated_in_mem_size())
-                        .sum::<u64>();
                     lm.frozen_layers.front().cloned()
                     // drop 'layers' lock to allow concurrent reads and writes
                 };
                 let Some(layer_to_flush) = layer_to_flush else {
                     break Ok(());
                 };
-                if num_frozen_layers
-                    > std::cmp::max(
-                        self.get_compaction_threshold(),
-                        DEFAULT_COMPACTION_THRESHOLD,
-                    )
-                    && frozen_layer_total_size >= /* 128 MB */ 128000000
-                {
-                    tracing::warn!(
-                        "too many frozen layers: {num_frozen_layers} layers with estimated in-mem size of {frozen_layer_total_size} bytes",
-                    );
-                }
                 match self.flush_frozen_layer(layer_to_flush, ctx).await {
                     Ok(this_layer_to_lsn) => {
                         flushed_to_lsn = std::cmp::max(flushed_to_lsn, this_layer_to_lsn);
@@ -4127,7 +4090,6 @@ impl Timeline {
     ) -> Result<ImageLayerCreationOutcome, CreateImageLayersError> {
         // Metadata keys image layer creation.
         let mut reconstruct_state = ValuesReconstructState::default();
-        let begin = Instant::now();
         let data = self
             .get_vectored_impl(partition.clone(), lsn, &mut reconstruct_state, ctx)
             .await?;
@@ -4144,11 +4106,14 @@ impl Timeline {
             (new_data, total_kb_retrieved / 1024, total_keys_retrieved)
         };
         let delta_files_accessed = reconstruct_state.get_delta_layers_visited();
-        let elapsed = begin.elapsed();
 
         let trigger_generation = delta_files_accessed as usize >= MAX_AUX_FILE_V2_DELTAS;
-        info!(
-            "metadata key compaction: trigger_generation={trigger_generation}, delta_files_accessed={delta_files_accessed}, total_kb_retrieved={total_kb_retrieved}, total_keys_retrieved={total_keys_retrieved}, read_time={}s", elapsed.as_secs_f64()
+        debug!(
+            trigger_generation,
+            delta_files_accessed,
+            total_kb_retrieved,
+            total_keys_retrieved,
+            "generate metadata images"
         );
 
         if !trigger_generation && mode == ImageLayerCreationMode::Try {

pageserver/src/tenant/timeline/compaction.rs

@@ -4,7 +4,7 @@
 //!
 //! The old legacy algorithm is implemented directly in `timeline.rs`.
 
-use std::collections::{BinaryHeap, HashMap, HashSet};
+use std::collections::{BinaryHeap, HashSet};
 use std::ops::{Deref, Range};
 use std::sync::Arc;
 
@@ -56,7 +56,7 @@ use pageserver_api::value::Value;
 
 use utils::lsn::Lsn;
 
-use pageserver_compaction::helpers::{fully_contains, overlaps_with};
+use pageserver_compaction::helpers::overlaps_with;
 use pageserver_compaction::interface::*;
 
 use super::CompactionError;
@@ -64,23 +64,6 @@ use super::CompactionError;
 /// Maximum number of deltas before generating an image layer in bottom-most compaction.
 const COMPACTION_DELTA_THRESHOLD: usize = 5;
 
-pub struct GcCompactionJobDescription {
-    /// All layers to read in the compaction job
-    selected_layers: Vec<Layer>,
-    /// GC cutoff of the job
-    gc_cutoff: Lsn,
-    /// LSNs to retain for the job
-    retain_lsns_below_horizon: Vec<Lsn>,
-    /// Maximum layer LSN processed in this compaction
-    max_layer_lsn: Lsn,
-    /// Only compact layers overlapping with this range
-    compaction_key_range: Range<Key>,
-    /// When partial compaction is enabled, these layers need to be rewritten to ensure no overlap.
-    /// This field is here solely for debugging. The field will not be read once the compaction
-    /// description is generated.
-    rewrite_layers: Vec<Arc<PersistentLayerDesc>>,
-}
-
 /// The result of bottom-most compaction for a single key at each LSN.
 #[derive(Debug)]
 #[cfg_attr(test, derive(PartialEq))]
@@ -1739,8 +1722,7 @@ impl Timeline {
         flags: EnumSet<CompactFlags>,
         ctx: &RequestContext,
     ) -> anyhow::Result<()> {
-        self.partial_compact_with_gc(Key::MIN..Key::MAX, cancel, flags, ctx)
-            .await
+        self.partial_compact_with_gc(None, cancel, flags, ctx).await
     }
 
     /// An experimental compaction building block that combines compaction with garbage collection.
@@ -1750,15 +1732,12 @@ impl Timeline {
     /// layers and image layers, which generates image layers on the gc horizon, drop deltas below gc horizon,
     /// and create delta layers with all deltas >= gc horizon.
     ///
-    /// If `key_range` is provided, it will only compact the keys within the range, aka partial compaction.
-    /// Partial compaction will read and process all layers overlapping with the key range, even if it might
-    /// contain extra keys. After the gc-compaction phase completes, delta layers that are not fully contained
-    /// within the key range will be rewritten to ensure they do not overlap with the delta layers. Providing
-    /// Key::MIN..Key..MAX to the function indicates a full compaction, though technically, `Key::MAX` is not
-    /// part of the range.
+    /// If `key_range`, it will only compact the keys within the range, aka partial compaction. This functionality
+    /// is not complete yet, and if it is set, only image layers will be generated.
+    ///
     pub(crate) async fn partial_compact_with_gc(
         self: &Arc<Self>,
-        compaction_key_range: Range<Key>,
+        compaction_key_range: Option<Range<Key>>,
         cancel: &CancellationToken,
         flags: EnumSet<CompactFlags>,
         ctx: &RequestContext,
@@ -1783,8 +1762,9 @@ impl Timeline {
         .await?;
 
         let dry_run = flags.contains(CompactFlags::DryRun);
+        let partial_compaction = compaction_key_range.is_some();
 
-        if compaction_key_range == (Key::MIN..Key::MAX) {
+        if let Some(ref compaction_key_range) = compaction_key_range {
             info!("running enhanced gc bottom-most compaction, dry_run={dry_run}, compaction_key_range={}..{}", compaction_key_range.start, compaction_key_range.end);
         } else {
             info!("running enhanced gc bottom-most compaction, dry_run={dry_run}");
@@ -1800,7 +1780,7 @@ impl Timeline {
         // The layer selection has the following properties:
         // 1. If a layer is in the selection, all layers below it are in the selection.
         // 2. Inferred from (1), for each key in the layer selection, the value can be reconstructed only with the layers in the layer selection.
-        let job_desc = {
+        let (layer_selection, gc_cutoff, retain_lsns_below_horizon) = if !partial_compaction {
             let guard = self.layers.read().await;
             let layers = guard.layer_map()?;
             let gc_info = self.gc_info.read().unwrap();
@@ -1830,21 +1810,9 @@ impl Timeline {
             };
             // Then, pick all the layers that are below the max_layer_lsn. This is to ensure we can pick all single-key
             // layers to compact.
-            let mut rewrite_layers = Vec::new();
             for desc in layers.iter_historic_layers() {
-                if desc.get_lsn_range().end <= max_layer_lsn
-                    && overlaps_with(&desc.get_key_range(), &compaction_key_range)
-                {
-                    // If the layer overlaps with the compaction key range, we need to read it to obtain all keys within the range,
-                    // even if it might contain extra keys
+                if desc.get_lsn_range().end <= max_layer_lsn {
                     selected_layers.push(guard.get_from_desc(&desc));
-                    // If the layer is not fully contained within the key range, we need to rewrite it if it's a delta layer (it's fine
-                    // to overlap image layers)
-                    if desc.is_delta()
-                        && !fully_contains(&compaction_key_range, &desc.get_key_range())
-                    {
-                        rewrite_layers.push(desc);
-                    }
                 }
             }
             if selected_layers.is_empty() {
@@ -1852,59 +1820,82 @@ impl Timeline {
                 return Ok(());
             }
             retain_lsns_below_horizon.sort();
-            GcCompactionJobDescription {
-                selected_layers,
-                gc_cutoff,
-                retain_lsns_below_horizon,
-                max_layer_lsn,
-                compaction_key_range,
-                rewrite_layers,
+            (selected_layers, gc_cutoff, retain_lsns_below_horizon)
+        } else {
+            // In case of partial compaction, we currently only support generating image layers, and therefore,
+            // we pick all layers that are below the lowest retain_lsn and does not intersect with any of the layers.
+            let guard = self.layers.read().await;
+            let layers = guard.layer_map()?;
+            let gc_info = self.gc_info.read().unwrap();
+            let mut min_lsn = gc_info.cutoffs.select_min();
+            for (lsn, _, _) in &gc_info.retain_lsns {
+                if lsn < &min_lsn {
+                    min_lsn = *lsn;
+                }
             }
+            for lsn in gc_info.leases.keys() {
+                if lsn < &min_lsn {
+                    min_lsn = *lsn;
+                }
+            }
+            let mut selected_layers = Vec::new();
+            drop(gc_info);
+            // |-------| |-------| |-------|
+            // | Delta | | Delta | | Delta | -- min_lsn could be intersecting with the layers
+            // |-------| |-------| |-------| <- we want to pick all the layers below min_lsn, so that
+            // | Delta | | Delta | | Delta |    ...we can remove them after compaction
+            // |-------| |-------| |-------|
+            // Pick all the layers intersect or below the min_lsn, get the largest LSN in the selected layers.
+            let Some(compaction_key_range) = compaction_key_range.as_ref() else {
+                unreachable!()
+            };
+            for desc in layers.iter_historic_layers() {
+                if desc.get_lsn_range().end <= min_lsn
+                    && overlaps_with(&desc.key_range, compaction_key_range)
+                {
+                    selected_layers.push(guard.get_from_desc(&desc));
+                }
+            }
+            if selected_layers.is_empty() {
+                info!("no layers to compact with gc");
+                return Ok(());
+            }
+            (selected_layers, min_lsn, Vec::new())
         };
         let lowest_retain_lsn = if self.ancestor_timeline.is_some() {
+            if partial_compaction {
+                warn!("partial compaction cannot run on child branches (for now)");
+                return Ok(());
+            }
             Lsn(self.ancestor_lsn.0 + 1)
         } else {
-            let res = job_desc
-                .retain_lsns_below_horizon
+            let res = retain_lsns_below_horizon
                 .first()
                 .copied()
-                .unwrap_or(job_desc.gc_cutoff);
+                .unwrap_or(gc_cutoff);
             if cfg!(debug_assertions) {
                 assert_eq!(
                     res,
-                    job_desc
-                        .retain_lsns_below_horizon
+                    retain_lsns_below_horizon
                         .iter()
                         .min()
                         .copied()
-                        .unwrap_or(job_desc.gc_cutoff)
+                        .unwrap_or(gc_cutoff)
                 );
             }
             res
         };
         info!(
-            "picked {} layers for compaction ({} layers need rewriting) with max_layer_lsn={} gc_cutoff={} lowest_retain_lsn={}, key_range={}..{}",
-            job_desc.selected_layers.len(),
-            job_desc.rewrite_layers.len(),
-            job_desc.max_layer_lsn,
-            job_desc.gc_cutoff,
-            lowest_retain_lsn,
-            job_desc.compaction_key_range.start,
-            job_desc.compaction_key_range.end
+            "picked {} layers for compaction with gc_cutoff={} lowest_retain_lsn={}",
+            layer_selection.len(),
+            gc_cutoff,
+            lowest_retain_lsn
         );
 
-        for layer in &job_desc.selected_layers {
-            debug!("read layer: {}", layer.layer_desc().key());
-        }
-        for layer in &job_desc.rewrite_layers {
-            debug!("rewrite layer: {}", layer.key());
-        }
-
-        self.check_compaction_space(&job_desc.selected_layers)
-            .await?;
+        self.check_compaction_space(&layer_selection).await?;
 
         // Generate statistics for the compaction
-        for layer in &job_desc.selected_layers {
+        for layer in &layer_selection {
             let desc = layer.layer_desc();
             if desc.is_delta() {
                 stat.visit_delta_layer(desc.file_size());
@@ -1915,25 +1906,25 @@ impl Timeline {
 
         // Step 1: construct a k-merge iterator over all layers.
         // Also, verify if the layer map can be split by drawing a horizontal line at every LSN start/end split point.
-        let layer_names = job_desc
-            .selected_layers
+        let layer_names: Vec<crate::tenant::storage_layer::LayerName> = layer_selection
             .iter()
             .map(|layer| layer.layer_desc().layer_name())
             .collect_vec();
         if let Some(err) = check_valid_layermap(&layer_names) {
-            warn!("gc-compaction layer map check failed because {}, this is normal if partial compaction is not finished yet", err);
+            bail!("cannot run gc-compaction because {}", err);
         }
         // The maximum LSN we are processing in this compaction loop
-        let end_lsn = job_desc
-            .selected_layers
+        let end_lsn = layer_selection
             .iter()
             .map(|l| l.layer_desc().lsn_range.end)
             .max()
             .unwrap();
+        // We don't want any of the produced layers to cover the full key range (i.e., MIN..MAX) b/c it will then be recognized
+        // as an L0 layer.
         let mut delta_layers = Vec::new();
         let mut image_layers = Vec::new();
         let mut downloaded_layers = Vec::new();
-        for layer in &job_desc.selected_layers {
+        for layer in &layer_selection {
             let resident_layer = layer.download_and_keep_resident().await?;
             downloaded_layers.push(resident_layer);
         }
@@ -1952,8 +1943,8 @@ impl Timeline {
             dense_ks,
             sparse_ks,
         )?;
-
-        // Step 2: Produce images+deltas.
+        // Step 2: Produce images+deltas. TODO: ensure newly-produced delta does not overlap with other deltas.
+        // Data of the same key.
         let mut accumulated_values = Vec::new();
         let mut last_key: Option<Key> = None;
 
@@ -1965,7 +1956,10 @@ impl Timeline {
                     self.conf,
                     self.timeline_id,
                     self.tenant_shard_id,
-                    job_desc.compaction_key_range.start,
+                    compaction_key_range
+                        .as_ref()
+                        .map(|x| x.start)
+                        .unwrap_or(Key::MIN),
                     lowest_retain_lsn,
                     self.get_compaction_target_size(),
                     ctx,
@@ -1985,13 +1979,6 @@ impl Timeline {
         )
         .await?;
 
-        #[derive(Default)]
-        struct RewritingLayers {
-            before: Option<DeltaLayerWriter>,
-            after: Option<DeltaLayerWriter>,
-        }
-        let mut delta_layer_rewriters = HashMap::<Arc<PersistentLayerKey>, RewritingLayers>::new();
-
         /// Returns None if there is no ancestor branch. Throw an error when the key is not found.
         ///
         /// Currently, we always get the ancestor image for each key in the child branch no matter whether the image
@@ -2017,51 +2004,10 @@ impl Timeline {
         // the key and LSN range are determined. However, to keep things simple here, we still
         // create this writer, and discard the writer in the end.
 
-        while let Some(((key, lsn, val), desc)) = merge_iter.next_with_trace().await? {
+        while let Some((key, lsn, val)) = merge_iter.next().await? {
             if cancel.is_cancelled() {
                 return Err(anyhow!("cancelled")); // TODO: refactor to CompactionError and pass cancel error
             }
-            if !job_desc.compaction_key_range.contains(&key) {
-                if !desc.is_delta {
-                    continue;
-                }
-                let rewriter = delta_layer_rewriters.entry(desc.clone()).or_default();
-                let rewriter = if key < job_desc.compaction_key_range.start {
-                    if rewriter.before.is_none() {
-                        rewriter.before = Some(
-                            DeltaLayerWriter::new(
-                                self.conf,
-                                self.timeline_id,
-                                self.tenant_shard_id,
-                                desc.key_range.start,
-                                desc.lsn_range.clone(),
-                                ctx,
-                            )
-                            .await?,
-                        );
-                    }
-                    rewriter.before.as_mut().unwrap()
-                } else if key >= job_desc.compaction_key_range.end {
-                    if rewriter.after.is_none() {
-                        rewriter.after = Some(
-                            DeltaLayerWriter::new(
-                                self.conf,
-                                self.timeline_id,
-                                self.tenant_shard_id,
-                                job_desc.compaction_key_range.end,
-                                desc.lsn_range.clone(),
-                                ctx,
-                            )
-                            .await?,
-                        );
-                    }
-                    rewriter.after.as_mut().unwrap()
-                } else {
-                    unreachable!()
-                };
-                rewriter.put_value(key, lsn, val, ctx).await?;
-                continue;
-            }
             match val {
                 Value::Image(_) => stat.visit_image_key(&val),
                 Value::WalRecord(_) => stat.visit_wal_key(&val),
@@ -2072,27 +2018,35 @@ impl Timeline {
                 }
                 accumulated_values.push((key, lsn, val));
             } else {
-                let last_key: &mut Key = last_key.as_mut().unwrap();
-                stat.on_unique_key_visited(); // TODO: adjust statistics for partial compaction
-                let retention = self
-                    .generate_key_retention(
-                        *last_key,
-                        &accumulated_values,
-                        job_desc.gc_cutoff,
-                        &job_desc.retain_lsns_below_horizon,
-                        COMPACTION_DELTA_THRESHOLD,
-                        get_ancestor_image(self, *last_key, ctx).await?,
-                    )
-                    .await?;
-                retention
-                    .pipe_to(
-                        *last_key,
-                        &mut delta_layer_writer,
-                        image_layer_writer.as_mut(),
-                        &mut stat,
-                        ctx,
-                    )
-                    .await?;
+                let last_key = last_key.as_mut().unwrap();
+                stat.on_unique_key_visited();
+                let skip_adding_key = if let Some(ref compaction_key_range) = compaction_key_range {
+                    !compaction_key_range.contains(last_key)
+                } else {
+                    false
+                };
+                if !skip_adding_key {
+                    let retention = self
+                        .generate_key_retention(
+                            *last_key,
+                            &accumulated_values,
+                            gc_cutoff,
+                            &retain_lsns_below_horizon,
+                            COMPACTION_DELTA_THRESHOLD,
+                            get_ancestor_image(self, *last_key, ctx).await?,
+                        )
+                        .await?;
+                    // Put the image into the image layer. Currently we have a single big layer for the compaction.
+                    retention
+                        .pipe_to(
+                            *last_key,
+                            &mut delta_layer_writer,
+                            image_layer_writer.as_mut(),
+                            &mut stat,
+                            ctx,
+                        )
+                        .await?;
+                }
                 accumulated_values.clear();
                 *last_key = key;
                 accumulated_values.push((key, lsn, val));
@@ -2103,42 +2057,34 @@ impl Timeline {
         let last_key = last_key.expect("no keys produced during compaction");
         stat.on_unique_key_visited();
 
-        let retention = self
-            .generate_key_retention(
-                last_key,
-                &accumulated_values,
-                job_desc.gc_cutoff,
-                &job_desc.retain_lsns_below_horizon,
-                COMPACTION_DELTA_THRESHOLD,
-                get_ancestor_image(self, last_key, ctx).await?,
-            )
-            .await?;
-        retention
-            .pipe_to(
-                last_key,
-                &mut delta_layer_writer,
-                image_layer_writer.as_mut(),
-                &mut stat,
-                ctx,
-            )
-            .await?;
-        // end: move the above part to the loop body
-
-        let mut rewrote_delta_layers = Vec::new();
-        for (key, writers) in delta_layer_rewriters {
-            if let Some(delta_writer_before) = writers.before {
-                let (desc, path) = delta_writer_before
-                    .finish(job_desc.compaction_key_range.start, ctx)
-                    .await?;
-                let layer = Layer::finish_creating(self.conf, self, desc, &path)?;
-                rewrote_delta_layers.push(layer);
-            }
-            if let Some(delta_writer_after) = writers.after {
-                let (desc, path) = delta_writer_after.finish(key.key_range.end, ctx).await?;
-                let layer = Layer::finish_creating(self.conf, self, desc, &path)?;
-                rewrote_delta_layers.push(layer);
-            }
+        let skip_adding_key = if let Some(ref compaction_key_range) = compaction_key_range {
+            !compaction_key_range.contains(&last_key)
+        } else {
+            false
+        };
+        if !skip_adding_key {
+            let retention = self
+                .generate_key_retention(
+                    last_key,
+                    &accumulated_values,
+                    gc_cutoff,
+                    &retain_lsns_below_horizon,
+                    COMPACTION_DELTA_THRESHOLD,
+                    get_ancestor_image(self, last_key, ctx).await?,
+                )
+                .await?;
+            // Put the image into the image layer. Currently we have a single big layer for the compaction.
+            retention
+                .pipe_to(
+                    last_key,
+                    &mut delta_layer_writer,
+                    image_layer_writer.as_mut(),
+                    &mut stat,
+                    ctx,
+                )
+                .await?;
         }
+        // end: move the above part to the loop body
 
         let discard = |key: &PersistentLayerKey| {
             let key = key.clone();
@@ -2147,7 +2093,10 @@ impl Timeline {
 
         let produced_image_layers = if let Some(writer) = image_layer_writer {
             if !dry_run {
-                let end_key = job_desc.compaction_key_range.end;
+                let end_key = compaction_key_range
+                    .as_ref()
+                    .map(|x| x.end)
+                    .unwrap_or(Key::MAX);
                 writer
                     .finish_with_discard_fn(self, ctx, end_key, discard)
                     .await?
@@ -2168,8 +2117,10 @@ impl Timeline {
             Vec::new()
         };
 
-        // TODO: make image/delta/rewrote_delta layers generation atomic. At this point, we already generated resident layers, and if
-        // compaction is cancelled at this point, we might have some layers that are not cleaned up.
+        if partial_compaction && !produced_delta_layers.is_empty() {
+            bail!("implementation error: partial compaction should not be producing delta layers (for now)");
+        }
+
         let mut compact_to = Vec::new();
         let mut keep_layers = HashSet::new();
         let produced_delta_layers_len = produced_delta_layers.len();
@@ -2177,83 +2128,51 @@ impl Timeline {
         for action in produced_delta_layers {
             match action {
                 BatchWriterResult::Produced(layer) => {
-                    if cfg!(debug_assertions) {
-                        info!("produced delta layer: {}", layer.layer_desc().key());
-                    }
                     stat.produce_delta_layer(layer.layer_desc().file_size());
                     compact_to.push(layer);
                 }
                 BatchWriterResult::Discarded(l) => {
-                    if cfg!(debug_assertions) {
-                        info!("discarded delta layer: {}", l);
-                    }
                     keep_layers.insert(l);
                     stat.discard_delta_layer();
                 }
             }
         }
-        for layer in &rewrote_delta_layers {
-            debug!(
-                "produced rewritten delta layer: {}",
-                layer.layer_desc().key()
-            );
-        }
-        compact_to.extend(rewrote_delta_layers);
         for action in produced_image_layers {
             match action {
                 BatchWriterResult::Produced(layer) => {
-                    debug!("produced image layer: {}", layer.layer_desc().key());
                     stat.produce_image_layer(layer.layer_desc().file_size());
                     compact_to.push(layer);
                 }
                 BatchWriterResult::Discarded(l) => {
-                    debug!("discarded image layer: {}", l);
                     keep_layers.insert(l);
                     stat.discard_image_layer();
                 }
             }
         }
-
-        let mut layer_selection = job_desc.selected_layers;
-
-        // Partial compaction might select more data than it processes, e.g., if
-        // the compaction_key_range only partially overlaps:
-        //
-        //         [---compaction_key_range---]
-        //   [---A----][----B----][----C----][----D----]
-        //
-        // For delta layers, we will rewrite the layers so that it is cut exactly at
-        // the compaction key range, so we can always discard them. However, for image
-        // layers, as we do not rewrite them for now, we need to handle them differently.
-        // Assume image layers  A, B, C, D are all in the `layer_selection`.
-        //
-        // The created image layers contain whatever is needed from B, C, and from
-        // `----]` of A, and from  `[---` of D.
-        //
-        // In contrast, `[---A` and `D----]` have not been processed, so, we must
-        // keep that data.
-        //
-        // The solution for now is to keep A and D completely if they are image layers.
-        // (layer_selection is what we'll remove from the layer map, so, retain what
-        // is _not_ fully covered by compaction_key_range).
-        for layer in &layer_selection {
-            if !layer.layer_desc().is_delta() {
-                if !overlaps_with(
-                    &layer.layer_desc().key_range,
-                    &job_desc.compaction_key_range,
-                ) {
-                    bail!("violated constraint: image layer outside of compaction key range");
-                }
-                if !fully_contains(
-                    &job_desc.compaction_key_range,
-                    &layer.layer_desc().key_range,
-                ) {
-                    keep_layers.insert(layer.layer_desc().key());
-                }
-            }
-        }
-
+        let mut layer_selection = layer_selection;
         layer_selection.retain(|x| !keep_layers.contains(&x.layer_desc().key()));
+        if let Some(ref compaction_key_range) = compaction_key_range {
+            // Partial compaction might select more data than it processes, e.g., if
+            // the compaction_key_range only partially overlaps:
+            //
+            //         [---compaction_key_range---]
+            //   [---A----][----B----][----C----][----D----]
+            //
+            // A,B,C,D are all in the `layer_selection`. The created image layers contain
+            // whatever is needed from B, C, and from `----]` of A, and from  `[--` of D.
+            //
+            // In contrast, `[--A-` and `--D----]` have not been processed, so, we must
+            // keep that data.
+            //
+            // The solution for now is to keep A and D completely.
+            // (layer_selection is what we'll remove from the layer map, so,
+            //  retain what is _not_ fully covered by compaction_key_range).
+            layer_selection.retain(|x| {
+                let key_range = &x.layer_desc().key_range;
+                key_range.start >= compaction_key_range.start
+                    && key_range.end <= compaction_key_range.end
+            });
+        }
 
         info!(
             "gc-compaction statistics: {}",
@@ -2273,7 +2192,6 @@ impl Timeline {
 
         // Step 3: Place back to the layer map.
         {
-            // TODO: sanity check if the layer map is valid (i.e., should not have overlaps)
             let mut guard = self.layers.write().await;
             guard
                 .open_mut()?

pageserver/src/tenant/timeline/offload.rs

@@ -47,18 +47,21 @@ pub(crate) async fn offload_timeline(
     match is_archived {
         Some(true) => (),
         Some(false) => {
-            tracing::warn!("tried offloading a non-archived timeline");
+            tracing::warn!(?is_archived, "tried offloading a non-archived timeline");
             return Err(OffloadError::NotArchived);
         }
         None => {
             // This is legal: calls to this function can race with the timeline shutting down
-            tracing::info!("tried offloading a timeline whose remote storage is not initialized");
+            tracing::info!(
+                ?is_archived,
+                "tried offloading a timeline whose remote storage is not initialized"
+            );
             return Err(OffloadError::Cancelled);
         }
     }
 
     // Now that the Timeline is in Stopping state, request all the related tasks to shut down.
-    timeline.shutdown(super::ShutdownMode::Flush).await;
+    timeline.shutdown(super::ShutdownMode::Hard).await;
 
     // TODO extend guard mechanism above with method
     // to make deletions possible while offloading is in progress

pageserver/src/walredo/apply_neon.rs

@@ -253,10 +253,6 @@ pub(crate) fn apply_in_neon(
             use bytes::BufMut;
             if *will_init {
                 assert!(*clear, "init record must be clear to ensure correctness");
-                assert!(
-                    page.is_empty(),
-                    "init record must be the first entry to ensure correctness"
-                );
             }
             if *clear {
                 page.clear();

pgxn/neon/logical_replication_monitor.c

@@ -1,8 +1,7 @@
-#include <dirent.h>
 #include <limits.h>
 #include <string.h>
+#include <dirent.h>
 #include <signal.h>
-#include <sys/stat.h>
 
 #include "postgres.h"
 
@@ -22,35 +21,17 @@
 
 static int	logical_replication_max_snap_files = 300;
 
-/*
- * According to Chi (shyzh), the pageserver _should_ be good with 10 MB worth of
- * snapshot files. Let's use 8 MB since 8 is a power of 2.
- */
-static int	logical_replication_max_logicalsnapdir_size = 8000;
-
-/*
- * A primitive description of a logical snapshot file including the LSN of the
- * file and its size.
- */
-typedef struct SnapDesc {
-	XLogRecPtr	lsn;
-	off_t		sz;
-} SnapDesc;
-
 PGDLLEXPORT void LogicalSlotsMonitorMain(Datum main_arg);
 
-/*
- * Sorts an array of snapshot descriptors by their LSN.
- */
 static int
-SnapDescComparator(const void *a, const void *b)
+LsnDescComparator(const void *a, const void *b)
 {
-	const SnapDesc	*desc1 = a;
-	const SnapDesc	*desc2 = b;
+	XLogRecPtr	lsn1 = *((const XLogRecPtr *) a);
+	XLogRecPtr	lsn2 = *((const XLogRecPtr *) b);
 
-	if (desc1->lsn < desc2->lsn)
+	if (lsn1 < lsn2)
 		return 1;
-	else if (desc1->lsn == desc2->lsn)
+	else if (lsn1 == lsn2)
 		return 0;
 	else
 		return -1;
@@ -62,39 +43,28 @@ SnapDescComparator(const void *a, const void *b)
  * slots having lower restart_lsn should be dropped.
  */
 static XLogRecPtr
-get_snapshots_cutoff_lsn(void)
+get_num_snap_files_lsn_threshold(void)
 {
-/* PG 18 has a constant defined for this, PG_LOGICAL_SNAPSHOTS_DIR */
-#define SNAPDIR "pg_logical/snapshots"
-
 	DIR		   *dirdesc;
-	int			dirdesc_fd;
 	struct dirent *de;
-	size_t		snapshot_index = 0;
-	SnapDesc   *snapshot_descriptors;
-	size_t		descriptors_allocated = 1024;
-	XLogRecPtr	cutoff = 0;
-	off_t		logicalsnapdir_size = 0;
-	const int	logical_replication_max_logicalsnapdir_size_bytes = logical_replication_max_logicalsnapdir_size * 1000;
+	char	   *snap_path = "pg_logical/snapshots/";
+	int			lsns_allocated = 1024;
+	int			lsns_num = 0;
+	XLogRecPtr *lsns;
+	XLogRecPtr	cutoff;
 
-	if (logical_replication_max_snap_files < 0 && logical_replication_max_logicalsnapdir_size < 0)
+	if (logical_replication_max_snap_files < 0)
 		return 0;
 
-	snapshot_descriptors = palloc(sizeof(*snapshot_descriptors) * descriptors_allocated);
-
-	dirdesc = AllocateDir(SNAPDIR);
-	dirdesc_fd = dirfd(dirdesc);
-	if (dirdesc_fd == -1)
-		ereport(ERROR, errmsg("failed to get a file descriptor for " SNAPDIR ": %m"));
+	lsns = palloc(sizeof(XLogRecPtr) * lsns_allocated);
 
 	/* find all .snap files and get their lsns */
-	while ((de = ReadDir(dirdesc, SNAPDIR)) != NULL)
+	dirdesc = AllocateDir(snap_path);
+	while ((de = ReadDir(dirdesc, snap_path)) != NULL)
 	{
+		XLogRecPtr	lsn;
 		uint32		hi;
 		uint32		lo;
-		struct stat	st;
-		XLogRecPtr	lsn;
-		SnapDesc   *desc;
 
 		if (strcmp(de->d_name, ".") == 0 ||
 			strcmp(de->d_name, "..") == 0)
@@ -109,69 +79,28 @@ get_snapshots_cutoff_lsn(void)
 
 		lsn = ((uint64) hi) << 32 | lo;
 		elog(DEBUG5, "found snap file %X/%X", LSN_FORMAT_ARGS(lsn));
-
-		if (fstatat(dirdesc_fd, de->d_name, &st, 0) == -1)
-			ereport(ERROR, errmsg("failed to get the size of " SNAPDIR "/%s: %m", de->d_name));
-
-		if (descriptors_allocated == snapshot_index)
+		if (lsns_allocated == lsns_num)
 		{
-			descriptors_allocated *= 2;
-			snapshot_descriptors = repalloc(snapshot_descriptors, sizeof(*snapshot_descriptors) * descriptors_allocated);
+			lsns_allocated *= 2;
+			lsns = repalloc(lsns, sizeof(XLogRecPtr) * lsns_allocated);
 		}
-
-		desc = &snapshot_descriptors[snapshot_index++];
-		desc->lsn = lsn;
-		desc->sz = st.st_size;
+		lsns[lsns_num++] = lsn;
 	}
-
-	qsort(snapshot_descriptors, snapshot_index, sizeof(*snapshot_descriptors), SnapDescComparator);
-
-	/* Are there more snapshot files than specified? */
-	if (logical_replication_max_snap_files <= snapshot_index)
+	/* sort by lsn desc */
+	qsort(lsns, lsns_num, sizeof(XLogRecPtr), LsnDescComparator);
+	/* and take cutoff at logical_replication_max_snap_files */
+	if (logical_replication_max_snap_files > lsns_num)
+		cutoff = 0;
+	/* have less files than cutoff */
+	else
 	{
-		cutoff = snapshot_descriptors[logical_replication_max_snap_files - 1].lsn;
-		elog(LOG,
-			"ls_monitor: dropping logical slots with restart_lsn lower %X/%X, found %zu snapshot files, limit is %d",
-			LSN_FORMAT_ARGS(cutoff), snapshot_index, logical_replication_max_snap_files);
+		cutoff = lsns[logical_replication_max_snap_files - 1];
+		elog(LOG, "ls_monitor: dropping logical slots with restart_lsn lower %X/%X, found %d .snap files, limit is %d",
+			 LSN_FORMAT_ARGS(cutoff), lsns_num, logical_replication_max_snap_files);
 	}
-
-	/* Is the size of the logical snapshots directory larger than specified?
-	 *
-	 * It's possible we could hit both thresholds, so remove any extra files
-	 * first, and then truncate based on size of the remaining files.
-	 */
-	if (logicalsnapdir_size > logical_replication_max_logicalsnapdir_size_bytes)
-	{
-		/* Unfortunately, iterating the directory does not guarantee any order
-		 * so we can't cache an index in the preceding loop.
-		 */
-
-		off_t		sz;
-		const XLogRecPtr original = cutoff;
-
-		sz = snapshot_descriptors[0].sz;
-		for (size_t i = 1; i < logical_replication_max_snap_files; ++i)
-		{
-			if (sz > logical_replication_max_logicalsnapdir_size_bytes)
-			{
-				cutoff = snapshot_descriptors[i - 1].lsn;
-				break;
-			}
-
-			sz += snapshot_descriptors[i].sz;
-		}
-
-		if (cutoff != original)
-			elog(LOG, "ls_monitor: dropping logical slots with restart_lsn lower than %X/%X, " SNAPDIR " is larger than %d KB",
-					LSN_FORMAT_ARGS(cutoff), logical_replication_max_logicalsnapdir_size);
-	}
-
-	pfree(snapshot_descriptors);
+	pfree(lsns);
 	FreeDir(dirdesc);
-
 	return cutoff;
-
-#undef SNAPDIR
 }
 
 void
@@ -189,16 +118,6 @@ InitLogicalReplicationMonitor(void)
 							0,
 							NULL, NULL, NULL);
 
-	DefineCustomIntVariable(
-							"neon.logical_replication_max_logicalsnapdir_size",
-							"Maximum allowed size of the pg_logical/snapshots directory (KB). When exceeded, slots are dropped until the limit is met. -1 disables the limit.",
-							NULL,
-							&logical_replication_max_logicalsnapdir_size,
-							8000, -1, INT_MAX,
-							PGC_SIGHUP,
-							GUC_UNIT_KB,
-							NULL, NULL, NULL);
-
 	memset(&bgw, 0, sizeof(bgw));
 	bgw.bgw_flags = BGWORKER_SHMEM_ACCESS;
 	bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
@@ -243,7 +162,7 @@ LogicalSlotsMonitorMain(Datum main_arg)
 		 * If there are too many .snap files, just drop all logical slots to
 		 * prevent aux files bloat.
 		 */
-		cutoff_lsn = get_snapshots_cutoff_lsn();
+		cutoff_lsn = get_num_snap_files_lsn_threshold();
 		if (cutoff_lsn > 0)
 		{
 			for (int i = 0; i < max_replication_slots; i++)

proxy/Cargo.toml

@@ -60,7 +60,7 @@ prometheus.workspace = true
 rand.workspace = true
 regex.workspace = true
 remote_storage = { version = "0.1", path = "../libs/remote_storage/" }
-reqwest = { workspace = true, features = ["rustls-tls-native-roots"] }
+reqwest.workspace = true
 reqwest-middleware = { workspace = true, features = ["json"] }
 reqwest-retry.workspace = true
 reqwest-tracing.workspace = true

proxy/src/auth/backend/jwt.rs

@@ -7,11 +7,8 @@ use arc_swap::ArcSwapOption;
 use dashmap::DashMap;
 use jose_jwk::crypto::KeyInfo;
 use reqwest::{redirect, Client};
-use reqwest_retry::policies::ExponentialBackoff;
-use reqwest_retry::RetryTransientMiddleware;
 use serde::de::Visitor;
 use serde::{Deserialize, Deserializer};
-use serde_json::value::RawValue;
 use signature::Verifier;
 use thiserror::Error;
 use tokio::time::Instant;
@@ -19,7 +16,7 @@ use tokio::time::Instant;
 use crate::auth::backend::ComputeCredentialKeys;
 use crate::context::RequestMonitoring;
 use crate::control_plane::errors::GetEndpointJwksError;
-use crate::http::read_body_with_limit;
+use crate::http::parse_json_body_with_limit;
 use crate::intern::RoleNameInt;
 use crate::types::{EndpointId, RoleName};
 
@@ -31,10 +28,6 @@ const MAX_RENEW: Duration = Duration::from_secs(3600);
 const MAX_JWK_BODY_SIZE: usize = 64 * 1024;
 const JWKS_USER_AGENT: &str = "neon-proxy";
 
-const JWKS_CONNECT_TIMEOUT: Duration = Duration::from_secs(2);
-const JWKS_FETCH_TIMEOUT: Duration = Duration::from_secs(5);
-const JWKS_FETCH_RETRIES: u32 = 3;
-
 /// How to get the JWT auth rules
 pub(crate) trait FetchAuthRules: Clone + Send + Sync + 'static {
     fn fetch_auth_rules(
@@ -62,7 +55,7 @@ pub(crate) struct AuthRule {
 }
 
 pub struct JwkCache {
-    client: reqwest_middleware::ClientWithMiddleware,
+    client: reqwest::Client,
 
     map: DashMap<(EndpointId, RoleName), Arc<JwkCacheEntryLock>>,
 }
@@ -124,14 +117,6 @@ impl Default for JwkCacheEntryLock {
     }
 }
 
-#[derive(Deserialize)]
-struct JwkSet<'a> {
-    /// we parse into raw-value because not all keys in a JWKS are ones
-    /// we can parse directly, so we parse them lazily.
-    #[serde(borrow)]
-    keys: Vec<&'a RawValue>,
-}
-
 impl JwkCacheEntryLock {
     async fn acquire_permit<'a>(self: &'a Arc<Self>) -> JwkRenewalPermit<'a> {
         JwkRenewalPermit::acquire_permit(self).await
@@ -145,7 +130,7 @@ impl JwkCacheEntryLock {
         &self,
         _permit: JwkRenewalPermit<'_>,
         ctx: &RequestMonitoring,
-        client: &reqwest_middleware::ClientWithMiddleware,
+        client: &reqwest::Client,
         endpoint: EndpointId,
         auth_rules: &F,
     ) -> Result<Arc<JwkCacheEntry>, JwtError> {
@@ -169,73 +154,22 @@ impl JwkCacheEntryLock {
             let req = client.get(rule.jwks_url.clone());
             // TODO(conrad): eventually switch to using reqwest_middleware/`new_client_with_timeout`.
             // TODO(conrad): We need to filter out URLs that point to local resources. Public internet only.
-            match req.send().await.and_then(|r| {
-                r.error_for_status()
-                    .map_err(reqwest_middleware::Error::Reqwest)
-            }) {
+            match req.send().await.and_then(|r| r.error_for_status()) {
                 // todo: should we re-insert JWKs if we want to keep this JWKs URL?
                 // I expect these failures would be quite sparse.
                 Err(e) => tracing::warn!(url=?rule.jwks_url, error=?e, "could not fetch JWKs"),
                 Ok(r) => {
                     let resp: http::Response<reqwest::Body> = r.into();
-
-                    let bytes = match read_body_with_limit(resp.into_body(), MAX_JWK_BODY_SIZE)
-                        .await
+                    match parse_json_body_with_limit::<jose_jwk::JwkSet>(
+                        resp.into_body(),
+                        MAX_JWK_BODY_SIZE,
+                    )
+                    .await
                     {
-                        Ok(bytes) => bytes,
-                        Err(e) => {
-                            tracing::warn!(url=?rule.jwks_url, error=?e, "could not decode JWKs");
-                            continue;
-                        }
-                    };
-
-                    match serde_json::from_slice::<JwkSet>(&bytes) {
                         Err(e) => {
                             tracing::warn!(url=?rule.jwks_url, error=?e, "could not decode JWKs");
                         }
                         Ok(jwks) => {
-                            // size_of::<&RawValue>() == 16
-                            // size_of::<jose_jwk::Jwk>() == 288
-                            // better to not pre-allocate this as it might be pretty large - especially if it has many
-                            // keys we don't want or need.
-                            // trivial 'attack': `{"keys":[` + repeat(`0`).take(30000).join(`,`) + `]}`
-                            // this would consume 8MiB just like that!
-                            let mut keys = vec![];
-                            let mut failed = 0;
-                            for key in jwks.keys {
-                                match serde_json::from_str::<jose_jwk::Jwk>(key.get()) {
-                                    Ok(key) => {
-                                        // if `use` (called `cls` in rust) is specified to be something other than signing,
-                                        // we can skip storing it.
-                                        if key
-                                            .prm
-                                            .cls
-                                            .as_ref()
-                                            .is_some_and(|c| *c != jose_jwk::Class::Signing)
-                                        {
-                                            continue;
-                                        }
-
-                                        keys.push(key);
-                                    }
-                                    Err(e) => {
-                                        tracing::debug!(url=?rule.jwks_url, failed=?e, "could not decode JWK");
-                                        failed += 1;
-                                    }
-                                }
-                            }
-                            keys.shrink_to_fit();
-
-                            if failed > 0 {
-                                tracing::warn!(url=?rule.jwks_url, failed, "could not decode JWKs");
-                            }
-
-                            if keys.is_empty() {
-                                tracing::warn!(url=?rule.jwks_url, "no valid JWKs found inside the response body");
-                                continue;
-                            }
-
-                            let jwks = jose_jwk::JwkSet { keys };
                             key_sets.insert(
                                 rule.id,
                                 KeySet {
@@ -245,7 +179,7 @@ impl JwkCacheEntryLock {
                                 },
                             );
                         }
-                    };
+                    }
                 }
             }
         }
@@ -262,7 +196,7 @@ impl JwkCacheEntryLock {
     async fn get_or_update_jwk_cache<F: FetchAuthRules>(
         self: &Arc<Self>,
         ctx: &RequestMonitoring,
-        client: &reqwest_middleware::ClientWithMiddleware,
+        client: &reqwest::Client,
         endpoint: EndpointId,
         fetch: &F,
     ) -> Result<Arc<JwkCacheEntry>, JwtError> {
@@ -316,7 +250,7 @@ impl JwkCacheEntryLock {
         self: &Arc<Self>,
         ctx: &RequestMonitoring,
         jwt: &str,
-        client: &reqwest_middleware::ClientWithMiddleware,
+        client: &reqwest::Client,
         endpoint: EndpointId,
         role_name: &RoleName,
         fetch: &F,
@@ -435,19 +369,8 @@ impl Default for JwkCache {
         let client = Client::builder()
             .user_agent(JWKS_USER_AGENT)
             .redirect(redirect::Policy::none())
-            .tls_built_in_native_certs(true)
-            .connect_timeout(JWKS_CONNECT_TIMEOUT)
-            .timeout(JWKS_FETCH_TIMEOUT)
             .build()
-            .expect("client config should be valid");
-
-        // Retry up to 3 times with increasing intervals between attempts.
-        let retry_policy = ExponentialBackoff::builder().build_with_max_retries(JWKS_FETCH_RETRIES);
-
-        let client = reqwest_middleware::ClientBuilder::new(client)
-            .with(RetryTransientMiddleware::new_with_policy(retry_policy))
-            .build();
-
+            .expect("using &str and standard redirect::Policy");
         JwkCache {
             client,
             map: DashMap::default(),
@@ -1286,63 +1209,4 @@ X0n5X2/pBLJzxZc62ccvZYVnctBiFs6HbSnxpuMQCfkt/BcR/ttIepBQQIW86wHL
             }
         }
     }
-
-    #[tokio::test]
-    async fn check_jwk_keycloak_regression() {
-        let (rs, valid_jwk) = new_rsa_jwk(RS1, "rs1".into());
-        let valid_jwk = serde_json::to_value(valid_jwk).unwrap();
-
-        // This is valid, but we cannot parse it as we have no support for encryption JWKs, only signature based ones.
-        // This is taken directly from keycloak.
-        let invalid_jwk = serde_json::json! {
-            {
-                "kid": "U-Jc9xRli84eNqRpYQoIPF-GNuRWV3ZvAIhziRW2sbQ",
-                "kty": "RSA",
-                "alg": "RSA-OAEP",
-                "use": "enc",
-                "n": "yypYWsEKmM_wWdcPnSGLSm5ytw1WG7P7EVkKSulcDRlrM6HWj3PR68YS8LySYM2D9Z-79oAdZGKhIfzutqL8rK1vS14zDuPpAM-RWY3JuQfm1O_-1DZM8-07PmVRegP5KPxsKblLf_My8ByH6sUOIa1p2rbe2q_b0dSTXYu1t0dW-cGL5VShc400YymvTwpc-5uYNsaVxZajnB7JP1OunOiuCJ48AuVp3PqsLzgoXqlXEB1ZZdch3xT3bxaTtNruGvG4xmLZY68O_T3yrwTCNH2h_jFdGPyXdyZToCMSMK2qSbytlfwfN55pT9Vv42Lz1YmoB7XRjI9aExKPc5AxFw",
-                "e": "AQAB",
-                "x5c": [
-                    "MIICmzCCAYMCBgGS41E6azANBgkqhkiG9w0BAQsFADARMQ8wDQYDVQQDDAZtYXN0ZXIwHhcNMjQxMDMxMTYwMTQ0WhcNMzQxMDMxMTYwMzI0WjARMQ8wDQYDVQQDDAZtYXN0ZXIwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQDLKlhawQqYz/BZ1w+dIYtKbnK3DVYbs/sRWQpK6VwNGWszodaPc9HrxhLwvJJgzYP1n7v2gB1kYqEh/O62ovysrW9LXjMO4+kAz5FZjcm5B+bU7/7UNkzz7Ts+ZVF6A/ko/GwpuUt/8zLwHIfqxQ4hrWnatt7ar9vR1JNdi7W3R1b5wYvlVKFzjTRjKa9PClz7m5g2xpXFlqOcHsk/U66c6K4InjwC5Wnc+qwvOCheqVcQHVll1yHfFPdvFpO02u4a8bjGYtljrw79PfKvBMI0faH+MV0Y/Jd3JlOgIxIwrapJvK2V/B83nmlP1W/jYvPViagHtdGMj1oTEo9zkDEXAgMBAAEwDQYJKoZIhvcNAQELBQADggEBAECYX59+Q9v6c9sb6Q0/C6IgLWG2nVCgVE1YWwIzz+68WrhlmNCRuPjY94roB+tc2tdHbj+Nh3LMzJk7L1KCQoW1+LPK6A6E8W9ad0YPcuw8csV2pUA3+H56exQMH0fUAPQAU7tXWvnQ7otcpV1XA8afn/NTMTsnxi9mSkor8MLMYQ3aeRyh1+LAchHBthWiltqsSUqXrbJF59u5p0ghquuKcWR3TXsA7klGYBgGU5KAJifr9XT87rN0bOkGvbeWAgKvnQnjZwxdnLqTfp/pRY/PiJJHhgIBYPIA7STGnMPjmJ995i34zhnbnd8WHXJA3LxrIMqLW/l8eIdvtM1w8KI="
-                ],
-                "x5t": "QhfzMMnuAfkReTgZ1HtrfyOeeZs",
-                "x5t#S256": "cmHDUdKgLiRCEN28D5FBy9IJLFmR7QWfm77SLhGTCTU"
-            }
-        };
-
-        let jwks = serde_json::json! {{ "keys": [invalid_jwk, valid_jwk ] }};
-        let jwks_addr = jwks_server(move |path| match path {
-            "/" => Some(serde_json::to_vec(&jwks).unwrap()),
-            _ => None,
-        })
-        .await;
-
-        let role_name = RoleName::from("anonymous");
-        let role = RoleNameInt::from(&role_name);
-
-        let rules = vec![AuthRule {
-            id: "foo".to_owned(),
-            jwks_url: format!("http://{jwks_addr}/").parse().unwrap(),
-            audience: None,
-            role_names: vec![role],
-        }];
-
-        let fetch = Fetch(rules);
-        let jwk_cache = JwkCache::default();
-
-        let endpoint = EndpointId::from("ep");
-
-        let token = new_rsa_jwt("rs1".into(), rs);
-
-        jwk_cache
-            .check_jwt(
-                &RequestMonitoring::test(),
-                endpoint.clone(),
-                &role_name,
-                &fetch,
-                &token,
-            )
-            .await
-            .unwrap();
-    }
 }

proxy/src/http/mod.rs

@@ -6,6 +6,7 @@ pub mod health_server;
 
 use std::time::Duration;
 
+use anyhow::bail;
 use bytes::Bytes;
 use http::Method;
 use http_body_util::BodyExt;
@@ -15,7 +16,7 @@ use reqwest_middleware::RequestBuilder;
 pub(crate) use reqwest_middleware::{ClientWithMiddleware, Error};
 pub(crate) use reqwest_retry::policies::ExponentialBackoff;
 pub(crate) use reqwest_retry::RetryTransientMiddleware;
-use thiserror::Error;
+use serde::de::DeserializeOwned;
 
 use crate::metrics::{ConsoleRequest, Metrics};
 use crate::url::ApiUrl;
@@ -121,19 +122,10 @@ impl Endpoint {
     }
 }
 
-#[derive(Error, Debug)]
-pub(crate) enum ReadBodyError {
-    #[error("Content length exceeds limit of {limit} bytes")]
-    BodyTooLarge { limit: usize },
-
-    #[error(transparent)]
-    Read(#[from] reqwest::Error),
-}
-
-pub(crate) async fn read_body_with_limit(
+pub(crate) async fn parse_json_body_with_limit<D: DeserializeOwned>(
     mut b: impl Body<Data = Bytes, Error = reqwest::Error> + Unpin,
     limit: usize,
-) -> Result<Vec<u8>, ReadBodyError> {
+) -> anyhow::Result<D> {
     // We could use `b.limited().collect().await.to_bytes()` here
     // but this ends up being slightly more efficient as far as I can tell.
 
@@ -141,20 +133,20 @@ pub(crate) async fn read_body_with_limit(
     // in reqwest, this value is influenced by the Content-Length header.
     let lower_bound = match usize::try_from(b.size_hint().lower()) {
         Ok(bound) if bound <= limit => bound,
-        _ => return Err(ReadBodyError::BodyTooLarge { limit }),
+        _ => bail!("Content length exceeds limit of {limit} bytes"),
     };
     let mut bytes = Vec::with_capacity(lower_bound);
 
     while let Some(frame) = b.frame().await.transpose()? {
         if let Ok(data) = frame.into_data() {
             if bytes.len() + data.len() > limit {
-                return Err(ReadBodyError::BodyTooLarge { limit });
+                bail!("Content length exceeds limit of {limit} bytes")
             }
             bytes.extend_from_slice(&data);
         }
     }
 
-    Ok(bytes)
+    Ok(serde_json::from_slice::<D>(&bytes)?)
 }
 
 #[cfg(test)]

proxy/src/lib.rs

@@ -1,6 +1,12 @@
 // rustc lints/lint groups
 // https://doc.rust-lang.org/rustc/lints/groups.html
-#![deny(deprecated, future_incompatible, let_underscore, nonstandard_style)]
+#![deny(
+    deprecated,
+    future_incompatible,
+    let_underscore,
+    nonstandard_style,
+    rust_2024_compatibility
+)]
 #![warn(clippy::all, clippy::pedantic, clippy::cargo)]
 // List of denied lints from the clippy::restriction group.
 // https://rust-lang.github.io/rust-clippy/master/index.html#?groups=restriction

proxy/src/serverless/conn_pool_lib.rs

@@ -16,7 +16,8 @@ use super::http_conn_pool::ClientDataHttp;
 use super::local_conn_pool::ClientDataLocal;
 use crate::auth::backend::ComputeUserInfo;
 use crate::context::RequestMonitoring;
-use crate::control_plane::messages::{ColdStartInfo, MetricsAuxInfo};
+use crate::control_plane::messages::ColdStartInfo;
+use crate::control_plane::messages::MetricsAuxInfo;
 use crate::metrics::{HttpEndpointPoolsGuard, Metrics};
 use crate::types::{DbName, EndpointCacheKey, RoleName};
 use crate::usage_metrics::{Ids, MetricCounter, USAGE_METRICS};

proxy/src/serverless/http_conn_pool.rs

@@ -7,6 +7,7 @@ use hyper::client::conn::http2;
 use hyper_util::rt::{TokioExecutor, TokioIo};
 use parking_lot::RwLock;
 use rand::Rng;
+use std::result::Result::Ok;
 use tokio::net::TcpStream;
 use tracing::{debug, error, info, info_span, Instrument};
 

safekeeper/benches/receive_wal.rs

@@ -262,7 +262,7 @@ fn bench_wal_acceptor_throughput(c: &mut Criterion) {
 
                 // Send requests.
                 for req in reqgen {
-                    while reply_rx.try_recv().is_ok() {} // discard replies, to avoid blocking
+                    _ = reply_rx.try_recv(); // discard any replies, to avoid blocking
                     let msg = ProposerAcceptorMessage::AppendRequest(req);
                     msg_tx.send(msg).await.expect("send failed");
                 }

safekeeper/src/auth.rs

@@ -20,8 +20,7 @@ pub fn check_permission(claims: &Claims, tenant_id: Option<TenantId>) -> Result<
             | Scope::PageServerApi
             | Scope::GenerationsApi
             | Scope::Infra
-            | Scope::Scrubber
-            | Scope::ControllerPeer,
+            | Scope::Scrubber,
             _,
         ) => Err(AuthError(
             format!(

safekeeper/src/metrics.rs

@@ -55,7 +55,7 @@ pub static WRITE_WAL_SECONDS: Lazy<Histogram> = Lazy::new(|| {
 pub static FLUSH_WAL_SECONDS: Lazy<Histogram> = Lazy::new(|| {
     register_histogram!(
         "safekeeper_flush_wal_seconds",
-        "Seconds spent syncing WAL to a disk (excluding segment initialization)",
+        "Seconds spent syncing WAL to a disk",
         DISK_FSYNC_SECONDS_BUCKETS.to_vec()
     )
     .expect("Failed to register safekeeper_flush_wal_seconds histogram")
@@ -217,8 +217,7 @@ pub static WAL_RECEIVER_QUEUE_DEPTH: Lazy<Histogram> = Lazy::new(|| {
     let mut buckets = pow2_buckets(1, MSG_QUEUE_SIZE);
     buckets.insert(0, 0.0);
     buckets.insert(buckets.len() - 1, (MSG_QUEUE_SIZE - 1) as f64);
-    // TODO: tweak this.
-    assert!(buckets.len() <= 16, "too many histogram buckets");
+    assert!(buckets.len() <= 12, "too many histogram buckets");
 
     register_histogram!(
         "safekeeper_wal_receiver_queue_depth",

safekeeper/src/receive_wal.rs

@@ -7,15 +7,14 @@ use crate::metrics::{
     WAL_RECEIVERS, WAL_RECEIVER_QUEUE_DEPTH, WAL_RECEIVER_QUEUE_DEPTH_TOTAL,
     WAL_RECEIVER_QUEUE_SIZE_TOTAL,
 };
-use crate::safekeeper::{
-    AcceptorProposerMessage, AppendRequest, AppendRequestHeader, ProposerAcceptorMessage,
-    ServerInfo,
-};
+use crate::safekeeper::AcceptorProposerMessage;
+use crate::safekeeper::ProposerAcceptorMessage;
+use crate::safekeeper::ServerInfo;
 use crate::timeline::WalResidentTimeline;
 use crate::wal_service::ConnectionId;
 use crate::GlobalTimelines;
 use anyhow::{anyhow, Context};
-use bytes::{BufMut as _, Bytes, BytesMut};
+use bytes::BytesMut;
 use parking_lot::MappedMutexGuard;
 use parking_lot::Mutex;
 use parking_lot::MutexGuard;
@@ -207,8 +206,7 @@ impl Drop for WalReceiverGuard {
     }
 }
 
-// TODO: reconsider this.
-pub const MSG_QUEUE_SIZE: usize = 4096;
+pub const MSG_QUEUE_SIZE: usize = 256;
 pub const REPLY_QUEUE_SIZE: usize = 16;
 
 impl SafekeeperPostgresHandler {
@@ -486,9 +484,6 @@ const FLUSH_INTERVAL: Duration = Duration::from_secs(1);
 /// every 5 seconds, for 12 samples per poll. This will give a count of up to 12x active timelines.
 const METRICS_INTERVAL: Duration = Duration::from_secs(5);
 
-/// The AppendRequest buffer size.
-const APPEND_BUFFER_SIZE: usize = 1024 * 1024;
-
 /// Encapsulates a task which takes messages from msg_rx, processes and pushes
 /// replies to reply_tx.
 ///
@@ -535,9 +530,6 @@ impl WalAcceptor {
     async fn run(&mut self) -> anyhow::Result<()> {
         let walreceiver_guard = self.tli.get_walreceivers().register(self.conn_id);
 
-        // Buffer AppendRequests to submit them as a single large write.
-        let mut append_buf = BufferedAppendRequest::new(APPEND_BUFFER_SIZE);
-
         // Periodically flush the WAL and compute metrics.
         let mut flush_ticker = tokio::time::interval(FLUSH_INTERVAL);
         flush_ticker.set_missed_tick_behavior(MissedTickBehavior::Delay);
@@ -554,7 +546,7 @@ impl WalAcceptor {
                 // Process inbound message.
                 msg = self.msg_rx.recv() => {
                     // If disconnected, break to flush WAL and return.
-                    let Some(msg) = msg else {
+                    let Some(mut msg) = msg else {
                         break;
                     };
 
@@ -571,44 +563,11 @@ impl WalAcceptor {
                     // This batches multiple appends per fsync. If the channel is empty after
                     // sending the reply, we'll schedule an immediate flush.
                     if let ProposerAcceptorMessage::AppendRequest(append_request) = msg {
-                        // Try to batch multiple messages into a single large write.
-                        if !append_buf.is_empty() || !self.msg_rx.is_empty() {
-                            if append_buf.add(&append_request) {
-                                continue; // message buffered, go get next message
-                            }
-
-                            // Full buffer, write it and buffer this message for next iteration.
-                            dirty = true;
-                            let buf_req = append_buf.take().expect("empty buffer");
-                            let buf_msg = ProposerAcceptorMessage::NoFlushAppendRequest(buf_req);
-                            let reply = self.tli.process_msg(&buf_msg).await?;
-                            drop(buf_msg); // allow reusing buffer for add
-                            assert!(append_buf.add(&append_request), "empty buffer rejected msg");
-                            reply
-                        } else {
-                            dirty = true;
-                            let msg = ProposerAcceptorMessage::NoFlushAppendRequest(append_request);
-                            self.tli.process_msg(&msg).await?
-                        }
-                    } else {
-                        self.tli.process_msg(&msg).await?
+                        msg = ProposerAcceptorMessage::NoFlushAppendRequest(append_request);
+                        dirty = true;
                     }
-                }
 
-                // If there are no pending messages, write the append buffer.
-                //
-                // NB: we don't also flush the WAL here. Otherwise we can get into a regime where we
-                // quickly drain msg_rx and fsync before the sender is able to repopulate msg_rx.
-                // This happens consistently due to Tokio scheduling, leading to overeager fsyncing.
-                // Instead, we perform the write without fsyncing and give the sender a chance to
-                // get scheduled and populate msg_rx for the next iteration. If there are no further
-                // messages, the next iteration will flush the WAL.
-                _ = future::ready(()), if self.msg_rx.is_empty() && !append_buf.is_empty() => {
-                    dirty = true;
-                    let buf_req = append_buf.take().expect("empty buffer");
-                    self.tli
-                        .process_msg(&ProposerAcceptorMessage::NoFlushAppendRequest(buf_req))
-                        .await?
+                    self.tli.process_msg(&msg).await?
                 }
 
                 // While receiving AppendRequests, flush the WAL periodically and respond with an
@@ -620,11 +579,11 @@ impl WalAcceptor {
                         .await?
                 }
 
-                // If there are no pending messages, flush the WAL and append buffer immediately.
+                // If there are no pending messages, flush the WAL immediately.
                 //
                 // TODO: this should be done via flush_ticker.reset_immediately(), but that's always
                 // delayed by 1ms due to this bug: https://github.com/tokio-rs/tokio/issues/6866.
-                _ = future::ready(()), if self.msg_rx.is_empty() && dirty => {
+                _ = future::ready(()), if dirty && self.msg_rx.is_empty() => {
                     dirty = false;
                     flush_ticker.reset();
                     self.tli
@@ -668,115 +627,3 @@ impl Drop for WalAcceptor {
         }
     }
 }
-
-/// Buffers WAL data for multiple AppendRequests, to submit them as a single write.
-struct BufferedAppendRequest {
-    /// The buffer capacity.
-    capacity: usize,
-    /// The buffered header and WAL data.
-    buf: Option<(AppendRequestHeader, BytesMut)>,
-    /// A previous buffer that can be reused when the returned message is dropped.
-    reuse_buf: Option<Bytes>,
-    /// If an AppendRequest is larger than the buffer capacity (when empty), just stash it here to
-    /// avoid growing the buffer and copying it. This will be returned as-is.
-    large: Option<AppendRequest>,
-}
-
-impl BufferedAppendRequest {
-    /// Creates a new append request buffer with the given capacity.
-    fn new(capacity: usize) -> Self {
-        Self {
-            capacity,
-            buf: None,
-            reuse_buf: None,
-            large: None,
-        }
-    }
-
-    /// Adds the given append request to the buffer, if possible. Returns `false` if the message
-    /// can't be buffered, leaving self unmodified. An empty buffer will always accept a message.
-    ///
-    /// If the buffer is not empty, the message must have the same term and proposer and contiguous
-    /// `begin_lsn` and `end_lsn`. The buffer must have available capacity for the entire
-    /// `wal_data`. If the message is greater than an empty buffer's capacity, it is accepted but
-    /// simply stashed away in `large` without growing the buffer.
-    pub fn add(&mut self, msg: &AppendRequest) -> bool {
-        // If there is a stashed large message, reject further messages.
-        if self.large.is_some() {
-            return false;
-        }
-
-        // If there is no existing buffer, initialize one with the message.
-        let Some((ref mut h, ref mut wal_data)) = self.buf else {
-            // If the message is larger than the buffer capacity, just stash it instead of growing.
-            if msg.wal_data.len() > self.capacity {
-                assert!(self.large.is_none());
-                self.large = Some(msg.clone()); // clone is cheap with Bytes
-                return true;
-            }
-
-            // Reuse a previous buffer, if any, or allocate a new one.
-            //
-            // TODO: try_into_mut() is essentially runtime borrow checking. If AppendRequest used a
-            // normal Vec<u8> we could do compile-time borrow checking instead and avoid panic.
-            let mut wal_data = match self.reuse_buf.take() {
-                Some(reuse_buf) => match reuse_buf.try_into_mut() {
-                    Ok(mut reuse_buf) => {
-                        assert_eq!(reuse_buf.capacity(), self.capacity);
-                        reuse_buf.clear();
-                        reuse_buf
-                    }
-                    Err(_) => panic!("couldn't reuse buffer, still in use"),
-                },
-                None => BytesMut::with_capacity(self.capacity),
-            };
-            // Copy the append request into the buffer.
-            wal_data.put_slice(&msg.wal_data);
-            self.buf = Some((msg.h, wal_data));
-            return true;
-        };
-
-        // The messages must have the same term and proposer.
-        if h.term != msg.h.term || h.proposer_uuid != msg.h.proposer_uuid {
-            return false;
-        }
-        // The messages must be contiguous.
-        if h.end_lsn != msg.h.begin_lsn {
-            return false;
-        }
-        // The message must fit in the buffer.
-        if wal_data.len() + msg.wal_data.len() > self.capacity {
-            return false;
-        }
-
-        // Add the message to the buffer, bumping the commit and truncate LSNs. We assume that later
-        // messages have later commit/truncate LSNs.
-        h.end_lsn = msg.h.end_lsn;
-        h.commit_lsn = msg.h.commit_lsn;
-        h.truncate_lsn = msg.h.truncate_lsn;
-        wal_data.put_slice(&msg.wal_data);
-        true
-    }
-
-    /// Returns true if there is no buffered message.
-    fn is_empty(&self) -> bool {
-        self.buf.is_none() && self.large.is_none()
-    }
-
-    /// Takes the buffered AppendRequest (if any), leaving a None in its place.
-    ///
-    /// NB: The returned `wal_data` Bytes must be dropped before the next call to `add()`, in order
-    /// to reuse the buffer. This is basically runtime borrow checking, because of Bytes.
-    fn take(&mut self) -> Option<AppendRequest> {
-        // If there is a stashed large message, return it.
-        if let Some(large) = self.large.take() {
-            assert!(self.buf.is_none(), "both buf and large are set");
-            return Some(large);
-        }
-
-        let (h, wal_data) = self.buf.take()?;
-        let wal_data = wal_data.freeze();
-        self.reuse_buf = Some(wal_data.clone()); // keep a reference to the buffer
-        Some(AppendRequest { h, wal_data })
-    }
-}

safekeeper/src/safekeeper.rs

@@ -296,13 +296,12 @@ pub struct ProposerElected {
 
 /// Request with WAL message sent from proposer to safekeeper. Along the way it
 /// communicates commit_lsn.
-#[derive(Clone, Debug)]
+#[derive(Debug)]
 pub struct AppendRequest {
     pub h: AppendRequestHeader,
     pub wal_data: Bytes,
 }
-
-#[derive(Debug, Clone, Copy, Deserialize)]
+#[derive(Debug, Clone, Deserialize)]
 pub struct AppendRequestHeader {
     // safekeeper's current term; if it is higher than proposer's, the compute is out of date.
     pub term: Term,
@@ -980,8 +979,7 @@ where
             self.wal_store.flush_wal().await?;
         }
 
-        // Update commit_lsn. It will be flushed to the control file regularly by the timeline
-        // manager, off of the WAL ingest hot path.
+        // Update commit_lsn.
         if msg.h.commit_lsn != Lsn(0) {
             self.update_commit_lsn(msg.h.commit_lsn).await?;
         }
@@ -994,6 +992,15 @@ where
         self.state.inmem.peer_horizon_lsn =
             max(self.state.inmem.peer_horizon_lsn, msg.h.truncate_lsn);
 
+        // Update truncate and commit LSN in control file.
+        // To avoid negative impact on performance of extra fsync, do it only
+        // when commit_lsn delta exceeds WAL segment size.
+        if self.state.commit_lsn + (self.state.server.wal_seg_size as u64)
+            < self.state.inmem.commit_lsn
+        {
+            self.state.flush().await?;
+        }
+
         trace!(
             "processed AppendRequest of len {}, begin_lsn={}, end_lsn={:?}, commit_lsn={:?}, truncate_lsn={:?}, flushed={:?}",
             msg.wal_data.len(),
@@ -1167,7 +1174,7 @@ mod tests {
             proposer_uuid: [0; 16],
         };
         let mut append_request = AppendRequest {
-            h: ar_hdr,
+            h: ar_hdr.clone(),
             wal_data: Bytes::from_static(b"b"),
         };
 
@@ -1241,7 +1248,7 @@ mod tests {
             proposer_uuid: [0; 16],
         };
         let append_request = AppendRequest {
-            h: ar_hdr,
+            h: ar_hdr.clone(),
             wal_data: Bytes::from_static(b"b"),
         };
 
@@ -1249,7 +1256,7 @@ mod tests {
         sk.process_msg(&ProposerAcceptorMessage::AppendRequest(append_request))
             .await
             .unwrap();
-        let mut ar_hrd2 = ar_hdr;
+        let mut ar_hrd2 = ar_hdr.clone();
         ar_hrd2.begin_lsn = Lsn(4);
         ar_hrd2.end_lsn = Lsn(5);
         let append_request = AppendRequest {

safekeeper/src/state.rs

@@ -4,7 +4,6 @@
 use std::{cmp::max, ops::Deref};
 
 use anyhow::{bail, Result};
-use postgres_ffi::WAL_SEGMENT_SIZE;
 use safekeeper_api::models::TimelineTermBumpResponse;
 use serde::{Deserialize, Serialize};
 use utils::{
@@ -145,7 +144,7 @@ impl TimelinePersistentState {
             ServerInfo {
                 pg_version: 170000, /* Postgres server version (major * 10000) */
                 system_id: 0,       /* Postgres system identifier */
-                wal_seg_size: WAL_SEGMENT_SIZE as u32,
+                wal_seg_size: 16 * 1024 * 1024,
             },
             vec![],
             Lsn::INVALID,

safekeeper/src/timeline.rs

@@ -108,11 +108,16 @@ pub type ReadGuardSharedState<'a> = RwLockReadGuard<'a, SharedState>;
 pub struct WriteGuardSharedState<'a> {
     tli: Arc<Timeline>,
     guard: RwLockWriteGuard<'a, SharedState>,
+    skip_update: bool,
 }
 
 impl<'a> WriteGuardSharedState<'a> {
     fn new(tli: Arc<Timeline>, guard: RwLockWriteGuard<'a, SharedState>) -> Self {
-        WriteGuardSharedState { tli, guard }
+        WriteGuardSharedState {
+            tli,
+            guard,
+            skip_update: false,
+        }
     }
 }
 
@@ -154,10 +159,12 @@ impl Drop for WriteGuardSharedState<'_> {
             }
         });
 
-        // send notification about shared state update
-        self.tli.shared_state_version_tx.send_modify(|old| {
-            *old += 1;
-        });
+        if !self.skip_update {
+            // send notification about shared state update
+            self.tli.shared_state_version_tx.send_modify(|old| {
+                *old += 1;
+            });
+        }
     }
 }
 

safekeeper/src/timeline_manager.rs

@@ -515,12 +515,7 @@ impl Manager {
             return;
         }
 
-        if state.cfile_last_persist_at.elapsed() > self.conf.control_file_save_interval
-            // If the control file's commit_lsn lags more than one segment behind the current
-            // commit_lsn, flush immediately to limit recovery time in case of a crash. We don't do
-            // this on the WAL ingest hot path since it incurs fsync latency.
-            || state.commit_lsn.saturating_sub(state.cfile_commit_lsn).0 >= self.wal_seg_size as u64
-        {
+        if state.cfile_last_persist_at.elapsed() > self.conf.control_file_save_interval {
             let mut write_guard = self.tli.write_shared_state().await;
             // it should be done in the background because it blocks manager task, but flush() should
             // be fast enough not to be a problem now

safekeeper/src/wal_storage.rs

@@ -18,6 +18,7 @@ use std::cmp::{max, min};
 use std::future::Future;
 use std::io::{self, SeekFrom};
 use std::pin::Pin;
+use std::time::Duration;
 use tokio::fs::{self, remove_file, File, OpenOptions};
 use tokio::io::{AsyncRead, AsyncWriteExt};
 use tokio::io::{AsyncReadExt, AsyncSeekExt};
@@ -257,9 +258,6 @@ impl PhysicalStorage {
             // Try to open existing partial file
             Ok((file, true))
         } else {
-            let _timer = WAL_STORAGE_OPERATION_SECONDS
-                .with_label_values(&["initialize_segment"])
-                .start_timer();
             // Create and fill new partial file
             //
             // We're using fdatasync during WAL writing, so file size must not
@@ -277,13 +275,22 @@ impl PhysicalStorage {
             });
             file.set_len(self.wal_seg_size as u64).await?;
 
-            if let Err(e) = durable_rename(&tmp_path, &wal_file_partial_path, !self.no_sync).await {
-                // Probably rename succeeded, but fsync of it failed. Remove
-                // the file then to avoid using it.
-                remove_file(wal_file_partial_path)
-                    .await
-                    .or_else(utils::fs_ext::ignore_not_found)?;
-                return Err(e.into());
+            match durable_rename(&tmp_path, &wal_file_partial_path, !self.no_sync).await {
+                Ok(fsync_latencies) => {
+                    for latency in fsync_latencies {
+                        if latency != Duration::ZERO {
+                            self.metrics.observe_flush_seconds(latency.as_secs_f64());
+                        }
+                    }
+                }
+                Err(e) => {
+                    // Probably rename succeeded, but fsync of it failed. Remove
+                    // the file then to avoid using it.
+                    remove_file(wal_file_partial_path)
+                        .await
+                        .or_else(utils::fs_ext::ignore_not_found)?;
+                    return Err(e.into());
+                }
             }
             Ok((file, true))
         }

storage_controller/src/http.rs

@@ -1033,7 +1033,7 @@ async fn handle_update_preferred_azs(req: Request<Body>) -> Result<Response<Body
 }
 
 async fn handle_step_down(req: Request<Body>) -> Result<Response<Body>, ApiError> {
-    check_permissions(&req, Scope::ControllerPeer)?;
+    check_permissions(&req, Scope::Admin)?;
 
     let req = match maybe_forward(req).await {
         ForwardOutcome::Forwarded(res) => {

test_runner/fixtures/auth_tokens.py

@@ -45,4 +45,3 @@ class TokenScope(str, Enum):
     SAFEKEEPER_DATA = "safekeeperdata"
     TENANT = "tenant"
     SCRUBBER = "scrubber"
-    INFRA = "infra"

test_runner/fixtures/benchmark_fixture.py

@@ -80,13 +80,7 @@ class PgBenchRunResult:
     ):
         stdout_lines = stdout.splitlines()
 
-        number_of_clients = 0
-        number_of_threads = 0
-        number_of_transactions_actually_processed = 0
-        latency_average = 0.0
         latency_stddev = None
-        tps = 0.0
-        scale = 0
 
         # we know significant parts of these values from test input
         # but to be precise take them from output

test_runner/fixtures/compare_fixtures.py

@@ -8,7 +8,7 @@ from contextlib import _GeneratorContextManager, contextmanager
 
 # Type-related stuff
 from pathlib import Path
-from typing import TYPE_CHECKING, final
+from typing import TYPE_CHECKING
 
 import pytest
 from _pytest.fixtures import FixtureRequest
@@ -70,12 +70,12 @@ class PgCompare(ABC):
 
     @contextmanager
     @abstractmethod
-    def record_pageserver_writes(self, out_name: str) -> Iterator[None]:
+    def record_pageserver_writes(self, out_name: str):
         pass
 
     @contextmanager
     @abstractmethod
-    def record_duration(self, out_name: str) -> Iterator[None]:
+    def record_duration(self, out_name: str):
         pass
 
     @contextmanager
@@ -105,7 +105,6 @@ class PgCompare(ABC):
         return results
 
 
-@final
 class NeonCompare(PgCompare):
     """PgCompare interface for the neon stack."""
 
@@ -207,7 +206,6 @@ class NeonCompare(PgCompare):
         return self.zenbenchmark.record_duration(out_name)
 
 
-@final
 class VanillaCompare(PgCompare):
     """PgCompare interface for vanilla postgres."""
 
@@ -273,7 +271,6 @@ class VanillaCompare(PgCompare):
         return self.zenbenchmark.record_duration(out_name)
 
 
-@final
 class RemoteCompare(PgCompare):
     """PgCompare interface for a remote postgres instance."""
 

test_runner/fixtures/endpoint/http.py

@@ -46,8 +46,3 @@ class EndpointHttpClient(requests.Session):
         )
         res.raise_for_status()
         return res.json()
-
-    def metrics(self) -> str:
-        res = self.get(f"http://localhost:{self.port}/metrics")
-        res.raise_for_status()
-        return res.text

test_runner/fixtures/h2server.py

@@ -4,14 +4,11 @@ https://python-hyper.org/projects/hyper-h2/en/stable/asyncio-example.html
 auth-broker -> local-proxy needs a h2 connection, so we need a h2 server :)
 """
 
-from __future__ import annotations
-
 import asyncio
 import collections
 import io
 import json
 from collections.abc import AsyncIterable
-from typing import TYPE_CHECKING, final
 
 import pytest_asyncio
 from h2.config import H2Configuration
@@ -28,45 +25,34 @@ from h2.events import (
 )
 from h2.exceptions import ProtocolError, StreamClosedError
 from h2.settings import SettingCodes
-from typing_extensions import override
-
-if TYPE_CHECKING:
-    from typing import Any, Optional
-
 
 RequestData = collections.namedtuple("RequestData", ["headers", "data"])
 
 
-@final
 class H2Server:
-    def __init__(self, host: str, port: int) -> None:
+    def __init__(self, host, port) -> None:
         self.host = host
         self.port = port
 
 
-@final
 class H2Protocol(asyncio.Protocol):
     def __init__(self):
         config = H2Configuration(client_side=False, header_encoding="utf-8")
         self.conn = H2Connection(config=config)
-        self.transport: Optional[asyncio.Transport] = None
-        self.stream_data: dict[int, RequestData] = {}
-        self.flow_control_futures: dict[int, asyncio.Future[Any]] = {}
+        self.transport = None
+        self.stream_data = {}
+        self.flow_control_futures = {}
 
-    @override
-    def connection_made(self, transport: asyncio.BaseTransport):
-        assert isinstance(transport, asyncio.Transport)
+    def connection_made(self, transport: asyncio.Transport):  # type: ignore[override]
         self.transport = transport
         self.conn.initiate_connection()
         self.transport.write(self.conn.data_to_send())
 
-    @override
-    def connection_lost(self, exc: Optional[Exception]):
+    def connection_lost(self, _exc):
         for future in self.flow_control_futures.values():
             future.cancel()
         self.flow_control_futures = {}
 
-    @override
     def data_received(self, data: bytes):
         assert self.transport is not None
         try:
@@ -91,7 +77,7 @@ class H2Protocol(asyncio.Protocol):
                     self.window_updated(event.stream_id, event.delta)
                 elif isinstance(event, RemoteSettingsChanged):
                     if SettingCodes.INITIAL_WINDOW_SIZE in event.changed_settings:
-                        self.window_updated(0, 0)
+                        self.window_updated(None, 0)
 
                 self.transport.write(self.conn.data_to_send())
 
@@ -137,7 +123,7 @@ class H2Protocol(asyncio.Protocol):
         else:
             stream_data.data.write(data)
 
-    def stream_reset(self, stream_id: int):
+    def stream_reset(self, stream_id):
         """
         A stream reset was sent. Stop sending data.
         """
@@ -145,7 +131,7 @@ class H2Protocol(asyncio.Protocol):
             future = self.flow_control_futures.pop(stream_id)
             future.cancel()
 
-    async def send_data(self, data: bytes, stream_id: int):
+    async def send_data(self, data, stream_id):
         """
         Send data according to the flow control rules.
         """
@@ -175,7 +161,7 @@ class H2Protocol(asyncio.Protocol):
             self.transport.write(self.conn.data_to_send())
             data = data[chunk_size:]
 
-    async def wait_for_flow_control(self, stream_id: int):
+    async def wait_for_flow_control(self, stream_id):
         """
         Waits for a Future that fires when the flow control window is opened.
         """
@@ -183,7 +169,7 @@ class H2Protocol(asyncio.Protocol):
         self.flow_control_futures[stream_id] = f
         await f
 
-    def window_updated(self, stream_id: int, delta):
+    def window_updated(self, stream_id, delta):
         """
         A window update frame was received. Unblock some number of flow control
         Futures.

test_runner/fixtures/neon_fixtures.py

@@ -286,7 +286,7 @@ class PgProtocol:
         return self.safe_psql_many([query], **kwargs)[0]
 
     def safe_psql_many(
-        self, queries: Iterable[str], log_query: bool = True, **kwargs: Any
+        self, queries: Iterable[str], log_query=True, **kwargs: Any
     ) -> list[list[tuple[Any, ...]]]:
         """
         Execute queries against the node and return all rows.
@@ -306,7 +306,7 @@ class PgProtocol:
                         result.append(cur.fetchall())
         return result
 
-    def safe_psql_scalar(self, query: str, log_query: bool = True) -> Any:
+    def safe_psql_scalar(self, query, log_query=True) -> Any:
         """
         Execute query returning single row with single column.
         """
@@ -1065,9 +1065,6 @@ class NeonEnv:
                 "http_auth_type": http_auth_type,
                 # Default which can be overriden with `NeonEnvBuilder.pageserver_config_override`
                 "availability_zone": "us-east-2a",
-                # Disable pageserver disk syncs in tests: when running tests concurrently, this avoids
-                # the pageserver taking a long time to start up due to syncfs flushing other tests' data
-                "no_sync": True,
             }
             if self.pageserver_virtual_file_io_engine is not None:
                 ps_cfg["virtual_file_io_engine"] = self.pageserver_virtual_file_io_engine
@@ -1785,7 +1782,7 @@ class NeonStorageController(MetricsGetter, LogUtils):
         self.request(
             "PUT",
             f"{self.api}/control/v1/node/{node_id}/drain",
-            headers=self.headers(TokenScope.INFRA),
+            headers=self.headers(TokenScope.ADMIN),
         )
 
     def cancel_node_drain(self, node_id):
@@ -1793,7 +1790,7 @@ class NeonStorageController(MetricsGetter, LogUtils):
         self.request(
             "DELETE",
             f"{self.api}/control/v1/node/{node_id}/drain",
-            headers=self.headers(TokenScope.INFRA),
+            headers=self.headers(TokenScope.ADMIN),
         )
 
     def node_fill(self, node_id):
@@ -1801,7 +1798,7 @@ class NeonStorageController(MetricsGetter, LogUtils):
         self.request(
             "PUT",
             f"{self.api}/control/v1/node/{node_id}/fill",
-            headers=self.headers(TokenScope.INFRA),
+            headers=self.headers(TokenScope.ADMIN),
         )
 
     def cancel_node_fill(self, node_id):
@@ -1809,14 +1806,14 @@ class NeonStorageController(MetricsGetter, LogUtils):
         self.request(
             "DELETE",
             f"{self.api}/control/v1/node/{node_id}/fill",
-            headers=self.headers(TokenScope.INFRA),
+            headers=self.headers(TokenScope.ADMIN),
         )
 
     def node_status(self, node_id):
         response = self.request(
             "GET",
             f"{self.api}/control/v1/node/{node_id}",
-            headers=self.headers(TokenScope.INFRA),
+            headers=self.headers(TokenScope.ADMIN),
         )
         return response.json()
 
@@ -1832,7 +1829,7 @@ class NeonStorageController(MetricsGetter, LogUtils):
         response = self.request(
             "GET",
             f"{self.api}/control/v1/node",
-            headers=self.headers(TokenScope.INFRA),
+            headers=self.headers(TokenScope.ADMIN),
         )
         return response.json()
 
@@ -1860,7 +1857,7 @@ class NeonStorageController(MetricsGetter, LogUtils):
         shard_count: Optional[int] = None,
         shard_stripe_size: Optional[int] = None,
         tenant_config: Optional[dict[Any, Any]] = None,
-        placement_policy: Optional[Union[dict[Any, Any], str]] = None,
+        placement_policy: Optional[Union[dict[Any, Any] | str]] = None,
     ):
         """
         Use this rather than pageserver_api() when you need to include shard parameters

test_runner/fixtures/pageserver/allowed_errors.py

@@ -93,8 +93,6 @@ DEFAULT_PAGESERVER_ALLOWED_ERRORS = (
     ".*WARN.*path=/v1/utilization .*request was dropped before completing",
     # Can happen during shutdown
     ".*scheduling deletion on drop failed: queue is in state Stopped.*",
-    # Too many frozen layers error is normal during intensive benchmarks
-    ".*too many frozen layers.*",
 )
 
 

test_runner/fixtures/pageserver/http.py

@@ -316,7 +316,7 @@ class PageserverHttpClient(requests.Session, MetricsGetter):
     def tenant_location_conf(
         self,
         tenant_id: Union[TenantId, TenantShardId],
-        location_conf: dict[str, Any],
+        location_conf=dict[str, Any],
         flush_ms=None,
         lazy: Optional[bool] = None,
     ):

test_runner/fixtures/pageserver/utils.py

@@ -56,8 +56,6 @@ def wait_for_upload(
     lsn: Lsn,
 ):
     """waits for local timeline upload up to specified lsn"""
-
-    current_lsn = Lsn(0)
     for i in range(20):
         current_lsn = remote_consistent_lsn(pageserver_http, tenant, timeline)
         if current_lsn >= lsn:
@@ -205,8 +203,6 @@ def wait_for_last_record_lsn(
     lsn: Lsn,
 ) -> Lsn:
     """waits for pageserver to catch up to a certain lsn, returns the last observed lsn."""
-
-    current_lsn = Lsn(0)
     for i in range(1000):
         current_lsn = last_record_lsn(pageserver_http, tenant, timeline)
         if current_lsn >= lsn:

test_runner/fixtures/paths.py

@@ -112,7 +112,7 @@ def compatibility_snapshot_dir() -> Iterator[Path]:
 
 
 @pytest.fixture(scope="session")
-def compatibility_neon_binpath() -> Iterator[Optional[Path]]:
+def compatibility_neon_binpath() -> Optional[Iterator[Path]]:
     if os.getenv("REMOTE_ENV"):
         return
     comp_binpath = None
@@ -133,7 +133,7 @@ def pg_distrib_dir(base_dir: Path) -> Iterator[Path]:
 
 
 @pytest.fixture(scope="session")
-def compatibility_pg_distrib_dir() -> Iterator[Optional[Path]]:
+def compatibility_pg_distrib_dir() -> Optional[Iterator[Path]]:
     compat_distrib_dir = None
     if env_compat_postgres_bin := os.environ.get("COMPATIBILITY_POSTGRES_DISTRIB_DIR"):
         compat_distrib_dir = Path(env_compat_postgres_bin).resolve()

test_runner/fixtures/pg_version.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
 import enum
+import os
 from typing import TYPE_CHECKING
 
+import pytest
 from typing_extensions import override
 
 if TYPE_CHECKING:
@@ -16,15 +18,12 @@ This fixture is used to determine which version of Postgres to use for tests.
 
 # Inherit PgVersion from str rather than int to make it easier to pass as a command-line argument
 # TODO: use enum.StrEnum for Python >= 3.11
+@enum.unique
 class PgVersion(str, enum.Enum):
     V14 = "14"
     V15 = "15"
     V16 = "16"
     V17 = "17"
-
-    # Default Postgres Version for tests that don't really depend on Postgres itself
-    DEFAULT = V16
-
     # Instead of making version an optional parameter in methods, we can use this fake entry
     # to explicitly rely on the default server version (could be different from pg_version fixture value)
     NOT_SET = "<-POSTRGRES VERSION IS NOT SET->"
@@ -60,3 +59,27 @@ class PgVersion(str, enum.Enum):
         # Make mypy happy
         # See https://github.com/python/mypy/issues/3974
         return None
+
+
+DEFAULT_VERSION: PgVersion = PgVersion.V16
+
+
+def skip_on_postgres(version: PgVersion, reason: str):
+    return pytest.mark.skipif(
+        PgVersion(os.environ.get("DEFAULT_PG_VERSION", DEFAULT_VERSION)) is version,
+        reason=reason,
+    )
+
+
+def xfail_on_postgres(version: PgVersion, reason: str):
+    return pytest.mark.xfail(
+        PgVersion(os.environ.get("DEFAULT_PG_VERSION", DEFAULT_VERSION)) is version,
+        reason=reason,
+    )
+
+
+def run_only_on_default_postgres(reason: str):
+    return pytest.mark.skipif(
+        PgVersion(os.environ.get("DEFAULT_PG_VERSION", DEFAULT_VERSION)) is not DEFAULT_VERSION,
+        reason=reason,
+    )

test_runner/fixtures/utils.py

@@ -25,7 +25,6 @@ from fixtures.pageserver.common_types import (
     parse_delta_layer,
     parse_image_layer,
 )
-from fixtures.pg_version import PgVersion
 
 if TYPE_CHECKING:
     from collections.abc import Iterable
@@ -38,7 +37,6 @@ if TYPE_CHECKING:
 
 
 Fn = TypeVar("Fn", bound=Callable[..., Any])
-
 COMPONENT_BINARIES = {
     "storage_controller": ("storage_controller",),
     "storage_broker": ("storage_broker",),
@@ -521,7 +519,7 @@ def assert_pageserver_backups_equal(left: Path, right: Path, skip_files: set[str
     This is essentially:
 
     lines=$(comm -3 \
-        <(mkdir left  && cd left  && tar xf "$left"  && find . -type f -print0 | xargs sha256sum | sort -k2) \
+        <(mkdir left && cd left && tar xf "$left" && find . -type f -print0 | xargs sha256sum | sort -k2) \
         <(mkdir right && cd right && tar xf "$right" && find . -type f -print0 | xargs sha256sum | sort -k2) \
         | wc -l)
     [ "$lines" = "0" ]
@@ -645,40 +643,3 @@ def allpairs_versions():
         )
         ids.append(f"combination_{''.join(cur_id)}")
     return {"argnames": "combination", "argvalues": tuple(argvalues), "ids": ids}
-
-
-def skip_on_postgres(version: PgVersion, reason: str):
-    return pytest.mark.skipif(
-        PgVersion(os.getenv("DEFAULT_PG_VERSION", PgVersion.DEFAULT)) is version,
-        reason=reason,
-    )
-
-
-def xfail_on_postgres(version: PgVersion, reason: str):
-    return pytest.mark.xfail(
-        PgVersion(os.getenv("DEFAULT_PG_VERSION", PgVersion.DEFAULT)) is version,
-        reason=reason,
-    )
-
-
-def run_only_on_default_postgres(reason: str):
-    return pytest.mark.skipif(
-        PgVersion(os.getenv("DEFAULT_PG_VERSION", PgVersion.DEFAULT)) is not PgVersion.DEFAULT,
-        reason=reason,
-    )
-
-
-def skip_in_debug_build(reason: str):
-    return pytest.mark.skipif(
-        os.getenv("BUILD_TYPE", "debug") == "debug",
-        reason=reason,
-    )
-
-
-def skip_on_ci(reason: str):
-    # `CI` variable is always set to `true` on GitHub
-    # Ref: https://docs.github.com/en/actions/writing-workflows/choosing-what-your-workflow-does/store-information-in-variables#default-environment-variables
-    return pytest.mark.skipif(
-        os.getenv("CI", "false") == "true",
-        reason=reason,
-    )

test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import os
 from pathlib import Path
 from typing import TYPE_CHECKING
 
@@ -13,7 +14,7 @@ from fixtures.neon_fixtures import (
     PgBin,
     wait_for_last_flush_lsn,
 )
-from fixtures.utils import get_scale_for_db, humantime_to_ms, skip_on_ci
+from fixtures.utils import get_scale_for_db, humantime_to_ms
 
 from performance.pageserver.util import (
     setup_pageserver_with_tenants,
@@ -37,8 +38,9 @@ if TYPE_CHECKING:
 @pytest.mark.parametrize("pgbench_scale", [get_scale_for_db(200)])
 @pytest.mark.parametrize("n_tenants", [500])
 @pytest.mark.timeout(10000)
-@skip_on_ci(
-    "This test needs lot of resources and should run on dedicated HW, not in github action runners as part of CI"
+@pytest.mark.skipif(
+    os.getenv("CI", "false") == "true",
+    reason="This test needs lot of resources and should run on dedicated HW, not in github action runners as part of CI",
 )
 def test_pageserver_characterize_throughput_with_n_tenants(
     neon_env_builder: NeonEnvBuilder,
@@ -64,8 +66,9 @@ def test_pageserver_characterize_throughput_with_n_tenants(
 @pytest.mark.parametrize("n_clients", [1, 64])
 @pytest.mark.parametrize("n_tenants", [1])
 @pytest.mark.timeout(2400)
-@skip_on_ci(
-    "This test needs lot of resources and should run on dedicated HW, not in github action runners as part of CI"
+@pytest.mark.skipif(
+    os.getenv("CI", "false") == "true",
+    reason="This test needs lot of resources and should run on dedicated HW, not in github action runners as part of CI",
 )
 def test_pageserver_characterize_latencies_with_1_client_and_throughput_with_many_clients_one_tenant(
     neon_env_builder: NeonEnvBuilder,

test_runner/performance/test_copy.py

@@ -2,13 +2,11 @@ from __future__ import annotations
 
 from contextlib import closing
 from io import BufferedReader, RawIOBase
-from typing import Optional, final
+from typing import Optional
 
 from fixtures.compare_fixtures import PgCompare
-from typing_extensions import override
 
 
-@final
 class CopyTestData(RawIOBase):
     def __init__(self, rows: int):
         self.rows = rows
@@ -16,7 +14,6 @@ class CopyTestData(RawIOBase):
         self.linebuf: Optional[bytes] = None
         self.ptr = 0
 
-    @override
     def readable(self):
         return True
 

test_runner/performance/test_logical_replication.py

@@ -149,16 +149,12 @@ def test_subscriber_lag(
                 check_pgbench_still_running(pub_workload, "pub")
                 check_pgbench_still_running(sub_workload, "sub")
 
-                pub_conn = psycopg2.connect(pub_connstr)
-                sub_conn = psycopg2.connect(sub_connstr)
-                pub_conn.autocommit = True
-                sub_conn.autocommit = True
-
-                with pub_conn.cursor() as pub_cur, sub_conn.cursor() as sub_cur:
-                    lag = measure_logical_replication_lag(sub_cur, pub_cur)
-
-                pub_conn.close()
-                sub_conn.close()
+                with (
+                    psycopg2.connect(pub_connstr) as pub_conn,
+                    psycopg2.connect(sub_connstr) as sub_conn,
+                ):
+                    with pub_conn.cursor() as pub_cur, sub_conn.cursor() as sub_cur:
+                        lag = measure_logical_replication_lag(sub_cur, pub_cur)
 
                 log.info(f"Replica lagged behind master by {lag} seconds")
                 zenbenchmark.record("replica_lag", lag, "s", MetricReport.LOWER_IS_BETTER)
@@ -210,7 +206,6 @@ def test_publisher_restart(
     sub_conn = psycopg2.connect(sub_connstr)
     pub_conn.autocommit = True
     sub_conn.autocommit = True
-
     with pub_conn.cursor() as pub_cur, sub_conn.cursor() as sub_cur:
         pub_cur.execute("SELECT 1 FROM pg_catalog.pg_publication WHERE pubname = 'pub1'")
         pub_exists = len(pub_cur.fetchall()) != 0
@@ -227,7 +222,6 @@ def test_publisher_restart(
             sub_cur.execute(f"create subscription sub1 connection '{pub_connstr}' publication pub1")
 
         initial_sync_lag = measure_logical_replication_lag(sub_cur, pub_cur)
-
     pub_conn.close()
     sub_conn.close()
 
@@ -254,17 +248,12 @@ def test_publisher_restart(
                     ["pgbench", "-c10", pgbench_duration, "-Mprepared"],
                     env=pub_env,
                 )
-
-                pub_conn = psycopg2.connect(pub_connstr)
-                sub_conn = psycopg2.connect(sub_connstr)
-                pub_conn.autocommit = True
-                sub_conn.autocommit = True
-
-                with pub_conn.cursor() as pub_cur, sub_conn.cursor() as sub_cur:
-                    lag = measure_logical_replication_lag(sub_cur, pub_cur)
-
-                pub_conn.close()
-                sub_conn.close()
+                with (
+                    psycopg2.connect(pub_connstr) as pub_conn,
+                    psycopg2.connect(sub_connstr) as sub_conn,
+                ):
+                    with pub_conn.cursor() as pub_cur, sub_conn.cursor() as sub_cur:
+                        lag = measure_logical_replication_lag(sub_cur, pub_cur)
 
                 log.info(f"Replica lagged behind master by {lag} seconds")
                 zenbenchmark.record("replica_lag", lag, "s", MetricReport.LOWER_IS_BETTER)
@@ -299,56 +288,58 @@ def test_snap_files(
     env = benchmark_project_pub.pgbench_env
     connstr = benchmark_project_pub.connstr
 
-    conn = psycopg2.connect(connstr)
-    conn.autocommit = True
-
-    with conn.cursor() as cur:
-        cur.execute("SELECT rolsuper FROM pg_roles WHERE rolname = 'neondb_owner'")
-        is_super = cast("bool", cur.fetchall()[0][0])
-        assert is_super, "This benchmark won't work if we don't have superuser"
-
-    conn.close()
+    with psycopg2.connect(connstr) as conn:
+        conn.autocommit = True
+        with conn.cursor() as cur:
+            cur.execute("SELECT rolsuper FROM pg_roles WHERE rolname = 'neondb_owner'")
+            is_super = cast("bool", cur.fetchall()[0][0])
+            assert is_super, "This benchmark won't work if we don't have superuser"
 
     pg_bin.run_capture(["pgbench", "-i", "-I", "dtGvp", "-s100"], env=env)
 
     conn = psycopg2.connect(connstr)
     conn.autocommit = True
+    cur = conn.cursor()
+    cur.execute("ALTER SYSTEM SET neon.logical_replication_max_snap_files = -1")
 
-    with conn.cursor() as cur:
-        cur.execute(
+    with psycopg2.connect(connstr) as conn:
+        conn.autocommit = True
+        with conn.cursor() as cur:
+            cur.execute("SELECT pg_reload_conf()")
+
+    with psycopg2.connect(connstr) as conn:
+        conn.autocommit = True
+        with conn.cursor() as cur:
+            cur.execute(
+                """
+                DO $$
+                    BEGIN
+                    IF EXISTS (
+                        SELECT 1
+                        FROM pg_replication_slots
+                        WHERE slot_name = 'slotter'
+                    ) THEN
+                        PERFORM pg_drop_replication_slot('slotter');
+                    END IF;
+                END $$;
             """
-            DO $$
-                BEGIN
-                IF EXISTS (
-                    SELECT 1
-                    FROM pg_replication_slots
-                    WHERE slot_name = 'slotter'
-                ) THEN
-                    PERFORM pg_drop_replication_slot('slotter');
-                END IF;
-            END $$;
-        """
-        )
-        cur.execute("SELECT pg_create_logical_replication_slot('slotter', 'test_decoding')")
-
-    conn.close()
+            )
+            cur.execute("SELECT pg_create_logical_replication_slot('slotter', 'test_decoding')")
 
     workload = pg_bin.run_nonblocking(["pgbench", "-c10", pgbench_duration, "-Mprepared"], env=env)
     try:
         start = time.time()
         prev_measurement = time.time()
         while time.time() - start < test_duration_min * 60:
-            conn = psycopg2.connect(connstr)
-            conn.autocommit = True
-
-            with conn.cursor() as cur:
-                cur.execute(
-                    "SELECT count(*) FROM (SELECT pg_log_standby_snapshot() FROM generate_series(1, 10000) g) s"
-                )
-                check_pgbench_still_running(workload)
-                cur.execute("SELECT pg_replication_slot_advance('slotter', pg_current_wal_lsn())")
-
-            conn.close()
+            with psycopg2.connect(connstr) as conn:
+                with conn.cursor() as cur:
+                    cur.execute(
+                        "SELECT count(*) FROM (SELECT pg_log_standby_snapshot() FROM generate_series(1, 10000) g) s"
+                    )
+                    check_pgbench_still_running(workload)
+                    cur.execute(
+                        "SELECT pg_replication_slot_advance('slotter', pg_current_wal_lsn())"
+                    )
 
             # Measure storage
             if time.time() - prev_measurement > test_interval_min * 60:

test_runner/performance/test_physical_replication.py

@@ -102,21 +102,15 @@ def test_ro_replica_lag(
                     check_pgbench_still_running(master_workload)
                     check_pgbench_still_running(replica_workload)
                     time.sleep(sync_interval_min * 60)
-
-                    conn_master = psycopg2.connect(master_connstr)
-                    conn_replica = psycopg2.connect(replica_connstr)
-                    conn_master.autocommit = True
-                    conn_replica.autocommit = True
-
                     with (
-                        conn_master.cursor() as cur_master,
-                        conn_replica.cursor() as cur_replica,
+                        psycopg2.connect(master_connstr) as conn_master,
+                        psycopg2.connect(replica_connstr) as conn_replica,
                     ):
-                        lag = measure_replication_lag(cur_master, cur_replica)
-
-                    conn_master.close()
-                    conn_replica.close()
-
+                        with (
+                            conn_master.cursor() as cur_master,
+                            conn_replica.cursor() as cur_replica,
+                        ):
+                            lag = measure_replication_lag(cur_master, cur_replica)
                     log.info(f"Replica lagged behind master by {lag} seconds")
                     zenbenchmark.record("replica_lag", lag, "s", MetricReport.LOWER_IS_BETTER)
             finally:
@@ -225,15 +219,11 @@ def test_replication_start_stop(
         pg_bin.run_capture(["pgbench", "-i", "-I", "dtGvp", "-s10"], env=master_env)
 
         # Sync replicas
-        conn_master = psycopg2.connect(master_connstr)
-        conn_master.autocommit = True
-
-        with conn_master.cursor() as cur_master:
-            for i in range(num_replicas):
-                conn_replica = psycopg2.connect(replica_connstr[i])
-                measure_replication_lag(cur_master, conn_replica.cursor())
-
-        conn_master.close()
+        with psycopg2.connect(master_connstr) as conn_master:
+            with conn_master.cursor() as cur_master:
+                for i in range(num_replicas):
+                    conn_replica = psycopg2.connect(replica_connstr[i])
+                    measure_replication_lag(cur_master, conn_replica.cursor())
 
         master_pgbench = pg_bin.run_nonblocking(
             [
@@ -287,22 +277,17 @@ def test_replication_start_stop(
 
             time.sleep(configuration_test_time_sec)
 
-            conn_master = psycopg2.connect(master_connstr)
-            conn_master.autocommit = True
-
-            with conn_master.cursor() as cur_master:
-                for ireplica in range(num_replicas):
-                    replica_conn = psycopg2.connect(replica_connstr[ireplica])
-                    lag = measure_replication_lag(cur_master, replica_conn.cursor())
-                    zenbenchmark.record(
-                        f"Replica {ireplica} lag", lag, "s", MetricReport.LOWER_IS_BETTER
-                    )
-                    log.info(
-                        f"Replica {ireplica} lagging behind master by {lag} seconds after configuration {iconfig:>b}"
-                    )
-
-            conn_master.close()
-
+            with psycopg2.connect(master_connstr) as conn_master:
+                with conn_master.cursor() as cur_master:
+                    for ireplica in range(num_replicas):
+                        replica_conn = psycopg2.connect(replica_connstr[ireplica])
+                        lag = measure_replication_lag(cur_master, replica_conn.cursor())
+                        zenbenchmark.record(
+                            f"Replica {ireplica} lag", lag, "s", MetricReport.LOWER_IS_BETTER
+                        )
+                        log.info(
+                            f"Replica {ireplica} lagging behind master by {lag} seconds after configuration {iconfig:>b}"
+                        )
         master_pgbench.terminate()
     except Exception as e:
         error_occurred = True

test_runner/regress/test_branch_and_gc.py

@@ -8,7 +8,7 @@ from fixtures.common_types import Lsn, TimelineId
 from fixtures.log_helper import log
 from fixtures.neon_fixtures import NeonEnv
 from fixtures.pageserver.http import TimelineCreate406
-from fixtures.utils import query_scalar, skip_in_debug_build
+from fixtures.utils import query_scalar
 
 
 # Test the GC implementation when running with branching.
@@ -48,8 +48,10 @@ from fixtures.utils import query_scalar, skip_in_debug_build
 # Because the delta layer D covering lsn1 is corrupted, creating a branch
 # starting from lsn1 should return an error as follows:
 #     could not find data for key ... at LSN ..., for request at LSN ...
-@skip_in_debug_build("times out in debug builds")
-def test_branch_and_gc(neon_simple_env: NeonEnv):
+def test_branch_and_gc(neon_simple_env: NeonEnv, build_type: str):
+    if build_type == "debug":
+        pytest.skip("times out in debug builds")
+
     env = neon_simple_env
     pageserver_http_client = env.pageserver.http_client()
 

test_runner/regress/test_compaction.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import enum
 import json
+import os
 import time
 from typing import TYPE_CHECKING
 
@@ -12,7 +13,7 @@ from fixtures.neon_fixtures import (
     generate_uploads_and_deletions,
 )
 from fixtures.pageserver.http import PageserverApiException
-from fixtures.utils import skip_in_debug_build, wait_until
+from fixtures.utils import wait_until
 from fixtures.workload import Workload
 
 if TYPE_CHECKING:
@@ -31,7 +32,7 @@ AGGRESIVE_COMPACTION_TENANT_CONF = {
 }
 
 
-@skip_in_debug_build("only run with release build")
+@pytest.mark.skipif(os.environ.get("BUILD_TYPE") == "debug", reason="only run with release build")
 def test_pageserver_compaction_smoke(neon_env_builder: NeonEnvBuilder):
     """
     This is a smoke test that compaction kicks in. The workload repeatedly churns

test_runner/regress/test_compute_locales.py

@@ -1,61 +0,0 @@
-from __future__ import annotations
-
-from typing import TYPE_CHECKING, cast
-
-from fixtures.pg_version import PgVersion
-
-if TYPE_CHECKING:
-    from collections.abc import Sequence
-
-    from fixtures.neon_fixtures import NeonEnv
-
-
-def test_default_locales(neon_simple_env: NeonEnv):
-    """
-    Test that the default locales for compute databases is C.UTF-8.
-    """
-    env = neon_simple_env
-
-    endpoint = env.endpoints.create_start("main")
-
-    domain_locales = cast(
-        "Sequence[str]",
-        endpoint.safe_psql(
-            "SELECT current_setting('lc_messages') AS lc_messages,"
-            + "current_setting('lc_monetary') AS lc_monetary,"
-            + "current_setting('lc_numeric') AS lc_numeric,"
-            + "current_setting('lc_time') AS lc_time"
-        )[0],
-    )
-    for dl in domain_locales:
-        assert dl == "C.UTF-8"
-
-    # Postgres 15 added the locale providers
-    if env.pg_version < PgVersion.V15:
-        results = cast(
-            "Sequence[str]",
-            endpoint.safe_psql(
-                "SELECT datcollate, datctype FROM pg_database WHERE datname = current_database()"
-            )[0],
-        )
-
-        datcollate = results[0]
-        datctype = results[1]
-    else:
-        results = cast(
-            "Sequence[str]",
-            endpoint.safe_psql(
-                "SELECT datlocprovider, datcollate, datctype FROM pg_database WHERE datname = current_database()"
-            )[0],
-        )
-        datlocprovider = results[0]
-        datcollate = results[1]
-        datctype = results[2]
-
-        if env.pg_version >= PgVersion.V17:
-            assert datlocprovider == "b", "The locale provider is not builtin"
-        else:
-            assert datlocprovider == "c", "The locale provider is not libc"
-
-    assert datcollate == "C.UTF-8"
-    assert datctype == "C.UTF-8"

test_runner/regress/test_download_extensions.py

@@ -12,7 +12,6 @@ from fixtures.neon_fixtures import (
     NeonEnvBuilder,
 )
 from fixtures.pg_version import PgVersion
-from fixtures.utils import skip_on_postgres
 from pytest_httpserver import HTTPServer
 from werkzeug.wrappers.request import Request
 from werkzeug.wrappers.response import Response
@@ -42,14 +41,17 @@ def neon_env_builder_local(
     return neon_env_builder
 
 
-@skip_on_postgres(PgVersion.V16, reason="TODO: PG16 extension building")
-@skip_on_postgres(PgVersion.V17, reason="TODO: PG17 extension building")
 def test_remote_extensions(
     httpserver: HTTPServer,
     neon_env_builder_local: NeonEnvBuilder,
     httpserver_listen_address,
     pg_version,
 ):
+    if pg_version == PgVersion.V16:
+        pytest.skip("TODO: PG16 extension building")
+    if pg_version == PgVersion.V17:
+        pytest.skip("TODO: PG17 extension building")
+
     # setup mock http server
     # that expects request for anon.tar.zst
     # and returns the requested file

test_runner/regress/test_ingestion_layer_size.py

@@ -4,22 +4,25 @@ from collections.abc import Iterable
 from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
+import pytest
 from fixtures.log_helper import log
 from fixtures.neon_fixtures import NeonEnvBuilder, wait_for_last_flush_lsn
 from fixtures.pageserver.http import HistoricLayerInfo, LayerMapInfo
-from fixtures.utils import human_bytes, skip_in_debug_build
+from fixtures.utils import human_bytes
 
 if TYPE_CHECKING:
     from typing import Union
 
 
-@skip_in_debug_build("debug run is unnecessarily slow")
-def test_ingesting_large_batches_of_images(neon_env_builder: NeonEnvBuilder):
+def test_ingesting_large_batches_of_images(neon_env_builder: NeonEnvBuilder, build_type: str):
     """
     Build a non-small GIN index which includes similarly batched up images in WAL stream as does pgvector
     to show that we no longer create oversized layers.
     """
 
+    if build_type == "debug":
+        pytest.skip("debug run is unnecessarily slow")
+
     minimum_initdb_size = 20 * 1024**2
     checkpoint_distance = 32 * 1024**2
     minimum_good_layer_size = checkpoint_distance * 0.9

test_runner/regress/test_installed_extensions.py

@@ -1,14 +1,6 @@
-from __future__ import annotations
-
-import time
 from logging import info
-from typing import TYPE_CHECKING
 
-from fixtures.log_helper import log
-from fixtures.metrics import parse_metrics
-
-if TYPE_CHECKING:
-    from fixtures.neon_fixtures import NeonEnv
+from fixtures.neon_fixtures import NeonEnv
 
 
 def test_installed_extensions(neon_simple_env: NeonEnv):
@@ -93,52 +85,3 @@ def test_installed_extensions(neon_simple_env: NeonEnv):
             assert ext["n_databases"] == 2
             ext["versions"].sort()
             assert ext["versions"] == ["1.2", "1.3"]
-
-    # check that /metrics endpoint is available
-    # ensure that we see the metric before and after restart
-    res = client.metrics()
-    info("Metrics: %s", res)
-    m = parse_metrics(res)
-    neon_m = m.query_all("installed_extensions", {"extension_name": "neon", "version": "1.2"})
-    assert len(neon_m) == 1
-    for sample in neon_m:
-        assert sample.value == 2
-    neon_m = m.query_all("installed_extensions", {"extension_name": "neon", "version": "1.3"})
-    assert len(neon_m) == 1
-    for sample in neon_m:
-        assert sample.value == 1
-
-    endpoint.stop()
-    endpoint.start()
-
-    timeout = 10
-    while timeout > 0:
-        try:
-            res = client.metrics()
-            timeout = -1
-            if len(parse_metrics(res).query_all("installed_extensions")) < 4:
-                # Assume that not all metrics that are collected yet
-                time.sleep(1)
-                timeout -= 1
-                continue
-        except Exception:
-            log.exception("failed to get metrics, assume they are not collected yet")
-            time.sleep(1)
-            timeout -= 1
-            continue
-
-        assert (
-            len(parse_metrics(res).query_all("installed_extensions")) >= 4
-        ), "Not all metrics are collected"
-
-        info("After restart metrics: %s", res)
-        m = parse_metrics(res)
-        neon_m = m.query_all("installed_extensions", {"extension_name": "neon", "version": "1.2"})
-        assert len(neon_m) == 1
-        for sample in neon_m:
-            assert sample.value == 1
-
-        neon_m = m.query_all("installed_extensions", {"extension_name": "neon", "version": "1.3"})
-        assert len(neon_m) == 1
-        for sample in neon_m:
-            assert sample.value == 1

test_runner/regress/test_layer_bloating.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import os
 
+import pytest
 from fixtures.log_helper import log
 from fixtures.neon_fixtures import (
     NeonEnvBuilder,
@@ -9,18 +10,12 @@ from fixtures.neon_fixtures import (
     wait_for_last_flush_lsn,
 )
 from fixtures.pg_version import PgVersion
-from fixtures.utils import skip_on_postgres
 
 
-@skip_on_postgres(
-    PgVersion.V14,
-    reason="pg_log_standby_snapshot() function is available since Postgres 16",
-)
-@skip_on_postgres(
-    PgVersion.V15,
-    reason="pg_log_standby_snapshot() function is available since Postgres 16",
-)
 def test_layer_bloating(neon_env_builder: NeonEnvBuilder, vanilla_pg):
+    if neon_env_builder.pg_version != PgVersion.V16:
+        pytest.skip("pg_log_standby_snapshot() function is available only in PG16")
+
     env = neon_env_builder.init_start(
         initial_tenant_conf={
             "gc_period": "0s",

test_runner/regress/test_layer_eviction.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import time
 
+import pytest
 from fixtures.log_helper import log
 from fixtures.neon_fixtures import (
     NeonEnvBuilder,
@@ -11,13 +12,17 @@ from fixtures.neon_fixtures import (
 from fixtures.pageserver.common_types import parse_layer_file_name
 from fixtures.pageserver.utils import wait_for_upload
 from fixtures.remote_storage import RemoteStorageKind
-from fixtures.utils import skip_in_debug_build
 
 
 # Crates a few layers, ensures that we can evict them (removing locally but keeping track of them anyway)
 # and then download them back.
-@skip_in_debug_build("times out in debug builds")
-def test_basic_eviction(neon_env_builder: NeonEnvBuilder):
+def test_basic_eviction(
+    neon_env_builder: NeonEnvBuilder,
+    build_type: str,
+):
+    if build_type == "debug":
+        pytest.skip("times out in debug builds")
+
     neon_env_builder.enable_pageserver_remote_storage(RemoteStorageKind.LOCAL_FS)
 
     env = neon_env_builder.init_start(

test_runner/regress/test_logging.py

@@ -5,7 +5,8 @@ import uuid
 import pytest
 from fixtures.log_helper import log
 from fixtures.neon_fixtures import NeonEnvBuilder
-from fixtures.utils import run_only_on_default_postgres, wait_until
+from fixtures.pg_version import run_only_on_default_postgres
+from fixtures.utils import wait_until
 
 
 @pytest.mark.parametrize("level", ["trace", "debug", "info", "warn", "error"])

test_runner/regress/test_logical_replication.py

@@ -4,31 +4,24 @@ import time
 from functools import partial
 from random import choice
 from string import ascii_lowercase
-from typing import TYPE_CHECKING, cast
 
-from fixtures.common_types import Lsn, TenantId, TimelineId
+from fixtures.common_types import Lsn
 from fixtures.log_helper import log
 from fixtures.neon_fixtures import (
+    NeonEnv,
+    NeonEnvBuilder,
+    PgProtocol,
     logical_replication_sync,
     wait_for_last_flush_lsn,
 )
 from fixtures.utils import wait_until
 
-if TYPE_CHECKING:
-    from fixtures.neon_fixtures import (
-        Endpoint,
-        NeonEnv,
-        NeonEnvBuilder,
-        PgProtocol,
-        VanillaPostgres,
-    )
-
 
 def random_string(n: int):
     return "".join([choice(ascii_lowercase) for _ in range(n)])
 
 
-def test_logical_replication(neon_simple_env: NeonEnv, vanilla_pg: VanillaPostgres):
+def test_logical_replication(neon_simple_env: NeonEnv, vanilla_pg):
     env = neon_simple_env
 
     tenant_id = env.initial_tenant
@@ -167,10 +160,10 @@ COMMIT;
 
 
 # Test that neon.logical_replication_max_snap_files works
-def test_obsolete_slot_drop(neon_simple_env: NeonEnv, vanilla_pg: VanillaPostgres):
-    def slot_removed(ep: Endpoint):
+def test_obsolete_slot_drop(neon_simple_env: NeonEnv, vanilla_pg):
+    def slot_removed(ep):
         assert (
-            ep.safe_psql(
+            endpoint.safe_psql(
                 "select count(*) from pg_replication_slots where slot_name = 'stale_slot'"
             )[0][0]
             == 0
@@ -261,7 +254,7 @@ FROM generate_series(1, 16384) AS seq; -- Inserts enough rows to exceed 16MB of
 
 
 # Tests that walsender correctly blocks until WAL is downloaded from safekeepers
-def test_lr_with_slow_safekeeper(neon_env_builder: NeonEnvBuilder, vanilla_pg: VanillaPostgres):
+def test_lr_with_slow_safekeeper(neon_env_builder: NeonEnvBuilder, vanilla_pg):
     neon_env_builder.num_safekeepers = 3
     env = neon_env_builder.init_start()
 
@@ -343,13 +336,13 @@ FROM generate_series(1, 16384) AS seq; -- Inserts enough rows to exceed 16MB of
 #
 # Most pages start with a contrecord, so we don't do anything special
 # to ensure that.
-def test_restart_endpoint(neon_simple_env: NeonEnv, vanilla_pg: VanillaPostgres):
+def test_restart_endpoint(neon_simple_env: NeonEnv, vanilla_pg):
     env = neon_simple_env
 
     env.create_branch("init")
     endpoint = env.endpoints.create_start("init")
-    tenant_id = TenantId(cast("str", endpoint.safe_psql("show neon.tenant_id")[0][0]))
-    timeline_id = TimelineId(cast("str", endpoint.safe_psql("show neon.timeline_id")[0][0]))
+    tenant_id = endpoint.safe_psql("show neon.tenant_id")[0][0]
+    timeline_id = endpoint.safe_psql("show neon.timeline_id")[0][0]
 
     cur = endpoint.connect().cursor()
     cur.execute("create table t(key int, value text)")
@@ -387,7 +380,7 @@ def test_restart_endpoint(neon_simple_env: NeonEnv, vanilla_pg: VanillaPostgres)
 # logical replication bug as such, but without logical replication,
 # records passed ot the WAL redo process are never large enough to hit
 # the bug.
-def test_large_records(neon_simple_env: NeonEnv, vanilla_pg: VanillaPostgres):
+def test_large_records(neon_simple_env: NeonEnv, vanilla_pg):
     env = neon_simple_env
 
     env.create_branch("init")
@@ -529,20 +522,15 @@ def logical_replication_wait_flush_lsn_sync(publisher: PgProtocol) -> Lsn:
     because for some WAL records like vacuum subscriber won't get any data at
     all.
     """
-    publisher_flush_lsn = Lsn(
-        cast("str", publisher.safe_psql("SELECT pg_current_wal_flush_lsn()")[0][0])
-    )
+    publisher_flush_lsn = Lsn(publisher.safe_psql("SELECT pg_current_wal_flush_lsn()")[0][0])
 
     def check_caughtup():
-        res = cast(
-            "tuple[str, str, str]",
-            publisher.safe_psql(
-                """
+        res = publisher.safe_psql(
+            """
 select sent_lsn, flush_lsn, pg_current_wal_flush_lsn() from pg_stat_replication sr, pg_replication_slots s
    where s.active_pid = sr.pid and s.slot_type = 'logical';
                                   """
-            )[0],
-        )
+        )[0]
         sent_lsn, flush_lsn, curr_publisher_flush_lsn = Lsn(res[0]), Lsn(res[1]), Lsn(res[2])
         log.info(
             f"sent_lsn={sent_lsn}, flush_lsn={flush_lsn}, publisher_flush_lsn={curr_publisher_flush_lsn}, waiting flush_lsn to reach {publisher_flush_lsn}"
@@ -557,7 +545,7 @@ select sent_lsn, flush_lsn, pg_current_wal_flush_lsn() from pg_stat_replication
 # flush_lsn reporting to publisher. Without this, subscriber may ack too far,
 # losing data on restart because publisher implicitly advances positition given
 # in START_REPLICATION to the confirmed_flush_lsn of the slot.
-def test_subscriber_synchronous_commit(neon_simple_env: NeonEnv, vanilla_pg: VanillaPostgres):
+def test_subscriber_synchronous_commit(neon_simple_env: NeonEnv, vanilla_pg):
     env = neon_simple_env
     # use vanilla as publisher to allow writes on it when safekeeper is down
     vanilla_pg.configure(
@@ -605,7 +593,7 @@ def test_subscriber_synchronous_commit(neon_simple_env: NeonEnv, vanilla_pg: Van
     # logical_replication_wait_flush_lsn_sync is expected to hang while
     # safekeeper is down.
     vanilla_pg.safe_psql("checkpoint;")
-    assert cast("int", sub.safe_psql_scalar("SELECT count(*) FROM t")) == 1000
+    assert sub.safe_psql_scalar("SELECT count(*) FROM t") == 1000
 
     # restart subscriber and ensure it can catch up lost tail again
     sub.stop(mode="immediate")

test_runner/regress/test_neon_cli.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import os
 import subprocess
 from pathlib import Path
 from typing import cast
@@ -14,7 +15,7 @@ from fixtures.neon_fixtures import (
     parse_project_git_version_output,
 )
 from fixtures.pageserver.http import PageserverHttpClient
-from fixtures.utils import run_only_on_default_postgres, skip_in_debug_build
+from fixtures.pg_version import PgVersion, skip_on_postgres
 
 
 def helper_compare_timeline_list(
@@ -194,8 +195,10 @@ def test_cli_start_stop_multi(neon_env_builder: NeonEnvBuilder):
     res.check_returncode()
 
 
-@run_only_on_default_postgres(reason="does not use postgres")
-@skip_in_debug_build("unit test for test support, either build works")
+@skip_on_postgres(PgVersion.V14, reason="does not use postgres")
+@pytest.mark.skipif(
+    os.environ.get("BUILD_TYPE") == "debug", reason="unit test for test support, either build works"
+)
 def test_parse_project_git_version_output_positive():
     commit = "b6f77b5816cf1dba12a3bc8747941182ce220846"
 
@@ -214,8 +217,10 @@ def test_parse_project_git_version_output_positive():
         assert parse_project_git_version_output(example) == commit
 
 
-@run_only_on_default_postgres(reason="does not use postgres")
-@skip_in_debug_build("unit test for test support, either build works")
+@skip_on_postgres(PgVersion.V14, reason="does not use postgres")
+@pytest.mark.skipif(
+    os.environ.get("BUILD_TYPE") == "debug", reason="unit test for test support, either build works"
+)
 def test_parse_project_git_version_output_local_docker():
     """
     Makes sure the tests don't accept the default version in Dockerfile one gets without providing
@@ -229,8 +234,10 @@ def test_parse_project_git_version_output_local_docker():
     assert input in str(e)
 
 
-@run_only_on_default_postgres(reason="does not use postgres")
-@skip_in_debug_build("unit test for test support, either build works")
+@skip_on_postgres(PgVersion.V14, reason="does not use postgres")
+@pytest.mark.skipif(
+    os.environ.get("BUILD_TYPE") == "debug", reason="cli api sanity, either build works"
+)
 def test_binaries_version_parses(neon_binpath: Path):
     """
     Ensures that we can parse the actual outputs of --version from a set of binaries.

test_runner/regress/test_pageserver_generations.py

@@ -656,7 +656,6 @@ def test_upgrade_generationless_local_file_paths(
     workload.write_rows(1000)
 
     attached_pageserver = env.get_tenant_pageserver(tenant_id)
-    assert attached_pageserver is not None
     secondary_pageserver = list([ps for ps in env.pageservers if ps.id != attached_pageserver.id])[
         0
     ]

test_runner/regress/test_pageserver_layer_rolling.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import os
 import time
 from typing import TYPE_CHECKING
 
@@ -15,7 +16,7 @@ from fixtures.neon_fixtures import (
 )
 from fixtures.pageserver.http import PageserverHttpClient
 from fixtures.pageserver.utils import wait_for_last_record_lsn, wait_for_upload
-from fixtures.utils import skip_in_debug_build, wait_until
+from fixtures.utils import wait_until
 
 if TYPE_CHECKING:
     from typing import Optional
@@ -226,9 +227,12 @@ def test_idle_checkpoints(neon_env_builder: NeonEnvBuilder):
         assert get_dirty_bytes(env) >= dirty_after_write
 
 
-# We have to use at least ~100MB of data to hit the lowest limit we can configure, which is
-# prohibitively slow in debug mode
-@skip_in_debug_build("Avoid running bulkier ingest tests in debug mode")
+@pytest.mark.skipif(
+    # We have to use at least ~100MB of data to hit the lowest limit we can configure, which is
+    # prohibitively slow in debug mode
+    os.getenv("BUILD_TYPE") == "debug",
+    reason="Avoid running bulkier ingest tests in debug mode",
+)
 def test_total_size_limit(neon_env_builder: NeonEnvBuilder):
     """
     Test that checkpoints are done based on total ephemeral layer size, even if no one timeline is

test_runner/regress/test_pageserver_restart.py

@@ -8,7 +8,7 @@ import pytest
 from fixtures.log_helper import log
 from fixtures.neon_fixtures import NeonEnvBuilder
 from fixtures.remote_storage import s3_storage
-from fixtures.utils import skip_in_debug_build, wait_until
+from fixtures.utils import wait_until
 
 
 # Test restarting page server, while safekeeper and compute node keep
@@ -155,8 +155,12 @@ def test_pageserver_restart(neon_env_builder: NeonEnvBuilder):
 # safekeeper and compute node keep running.
 @pytest.mark.timeout(540)
 @pytest.mark.parametrize("shard_count", [None, 4])
-@skip_in_debug_build("times out in debug builds")
-def test_pageserver_chaos(neon_env_builder: NeonEnvBuilder, shard_count: Optional[int]):
+def test_pageserver_chaos(
+    neon_env_builder: NeonEnvBuilder, build_type: str, shard_count: Optional[int]
+):
+    if build_type == "debug":
+        pytest.skip("times out in debug builds")
+
     # same rationale as with the immediate stop; we might leave orphan layers behind.
     neon_env_builder.disable_scrub_on_exit()
     neon_env_builder.enable_pageserver_remote_storage(s3_storage())

test_runner/regress/test_pageserver_secondary.py

@@ -17,7 +17,7 @@ from fixtures.pageserver.utils import (
     wait_for_upload_queue_empty,
 )
 from fixtures.remote_storage import LocalFsStorage, RemoteStorageKind, S3Storage, s3_storage
-from fixtures.utils import skip_in_debug_build, wait_until
+from fixtures.utils import wait_until
 from fixtures.workload import Workload
 from werkzeug.wrappers.request import Request
 from werkzeug.wrappers.response import Response
@@ -765,7 +765,7 @@ def test_secondary_background_downloads(neon_env_builder: NeonEnvBuilder):
     assert download_rate < expect_download_rate * 2
 
 
-@skip_in_debug_build("only run with release build")
+@pytest.mark.skipif(os.environ.get("BUILD_TYPE") == "debug", reason="only run with release build")
 @pytest.mark.parametrize("via_controller", [True, False])
 def test_slow_secondary_downloads(neon_env_builder: NeonEnvBuilder, via_controller: bool):
     """