Simplify low-hanging fruit

even though it's a std::io::Error

Co-authored-by: Christian Schwarz <christian@neon.tech>

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

@@ -1,184 +0,0 @@
-name: 'Create Allure report'
-description: 'Generate Allure report from uploaded by actions/allure-report-store tests results'
-
-outputs:
-  report-url:
-    description: 'Allure report URL'
-    value: ${{ steps.generate-report.outputs.report-url }}
-  report-json-url:
-    description: 'Allure report JSON URL'
-    value: ${{ steps.generate-report.outputs.report-json-url }}
-
-runs:
-  using: "composite"
-
-  steps:
-    # We're using some of env variables quite offen, so let's set them once.
-    #
-    # It would be nice to have them set in common runs.env[0] section, but it doesn't work[1]
-    #
-    # - [0] https://docs.github.com/en/actions/creating-actions/metadata-syntax-for-github-actions#runsenv
-    # - [1] https://github.com/neondatabase/neon/pull/3907#discussion_r1154703456
-    #
-    - name: Set variables
-      shell: bash -euxo pipefail {0}
-      run: |
-        PR_NUMBER=$(jq --raw-output .pull_request.number "$GITHUB_EVENT_PATH" || true)
-        if [ "${PR_NUMBER}" != "null" ]; then
-          BRANCH_OR_PR=pr-${PR_NUMBER}
-        elif [ "${GITHUB_REF_NAME}" = "main" ] || [ "${GITHUB_REF_NAME}" = "release" ]; then
-          # Shortcut for special branches
-          BRANCH_OR_PR=${GITHUB_REF_NAME}
-        else
-          BRANCH_OR_PR=branch-$(printf "${GITHUB_REF_NAME}" | tr -c "[:alnum:]._-" "-")
-        fi
-
-        LOCK_FILE=reports/${BRANCH_OR_PR}/lock.txt
-
-        WORKDIR=/tmp/${BRANCH_OR_PR}-$(date +%s)
-        mkdir -p ${WORKDIR}
-
-        echo "BRANCH_OR_PR=${BRANCH_OR_PR}" >> $GITHUB_ENV
-        echo "LOCK_FILE=${LOCK_FILE}"       >> $GITHUB_ENV
-        echo "WORKDIR=${WORKDIR}"           >> $GITHUB_ENV
-        echo "BUCKET=${BUCKET}"             >> $GITHUB_ENV
-      env:
-        BUCKET: neon-github-public-dev
-
-    # TODO: We can replace with a special docker image with Java and Allure pre-installed
-    - uses: actions/setup-java@v3
-      with:
-        distribution: 'temurin'
-        java-version: '17'
-
-    - name: Install Allure
-      shell: bash -euxo pipefail {0}
-      run: |
-        if ! which allure; then
-          ALLURE_ZIP=allure-${ALLURE_VERSION}.zip
-          wget -q https://github.com/allure-framework/allure2/releases/download/${ALLURE_VERSION}/${ALLURE_ZIP}
-          echo "${ALLURE_ZIP_MD5}  ${ALLURE_ZIP}" | md5sum -c
-          unzip -q ${ALLURE_ZIP}
-          echo "$(pwd)/allure-${ALLURE_VERSION}/bin" >> $GITHUB_PATH
-          rm -f ${ALLURE_ZIP}
-        fi
-      env:
-        ALLURE_VERSION: 2.22.0
-        ALLURE_ZIP_MD5: d5c9f0989b896482536956340a7d5ec9
-
-    # Potentially we could have several running build for the same key (for example, for the main branch), so we use improvised lock for this
-    - name: Acquire lock
-      shell: bash -euxo pipefail {0}
-      run: |
-        LOCK_TIMEOUT=300 # seconds
-
-        LOCK_CONTENT="${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
-        echo ${LOCK_CONTENT} > ${WORKDIR}/lock.txt
-
-        # Do it up to 5 times to avoid race condition
-        for _ in $(seq 1 5); do
-          for i in $(seq 1 ${LOCK_TIMEOUT}); do
-            LOCK_ACQUIRED=$(aws s3api head-object --bucket neon-github-public-dev --key ${LOCK_FILE} | jq --raw-output '.LastModified' || true)
-            # `date --date="..."` is supported only by gnu date (i.e. it doesn't work on BSD/macOS)
-            if [ -z "${LOCK_ACQUIRED}" ] || [ "$(( $(date +%s) - $(date --date="${LOCK_ACQUIRED}" +%s) ))" -gt "${LOCK_TIMEOUT}" ]; then
-              break
-            fi
-            sleep 1
-          done
-
-          aws s3 mv --only-show-errors ${WORKDIR}/lock.txt "s3://${BUCKET}/${LOCK_FILE}"
-
-          # Double-check that exactly THIS run has acquired the lock
-          aws s3 cp --only-show-errors "s3://${BUCKET}/${LOCK_FILE}" ./lock.txt
-          if [ "$(cat lock.txt)" = "${LOCK_CONTENT}" ]; then
-            break
-          fi
-        done
-
-    - name: Generate and publish final Allure report
-      id: generate-report
-      shell: bash -euxo pipefail {0}
-      run: |
-        REPORT_PREFIX=reports/${BRANCH_OR_PR}
-        RAW_PREFIX=reports-raw/${BRANCH_OR_PR}/${GITHUB_RUN_ID}
-
-        # Get previously uploaded data for this run
-        ZSTD_NBTHREADS=0
-
-        S3_FILEPATHS=$(aws s3api list-objects-v2 --bucket ${BUCKET} --prefix ${RAW_PREFIX}/ | jq --raw-output  '.Contents[].Key')
-        if [ -z "$S3_FILEPATHS" ]; then
-          # There's no previously uploaded data for this $GITHUB_RUN_ID
-          exit 0
-        fi
-        for S3_FILEPATH in ${S3_FILEPATHS}; do
-          time aws s3 cp --only-show-errors "s3://${BUCKET}/${S3_FILEPATH}" "${WORKDIR}"
-
-          archive=${WORKDIR}/$(basename $S3_FILEPATH)
-          mkdir -p ${archive%.tar.zst}
-          time tar -xf ${archive} -C ${archive%.tar.zst}
-          rm -f ${archive}
-        done
-
-        # Get history trend
-        time aws s3 cp --recursive --only-show-errors "s3://${BUCKET}/${REPORT_PREFIX}/latest/history" "${WORKDIR}/latest/history" || true
-
-        # Generate report
-        time allure generate --clean --output ${WORKDIR}/report ${WORKDIR}/*
-
-        # Replace a logo link with a redirect to the latest version of the report
-        sed -i 's|<a href="." class=|<a href="https://'${BUCKET}'.s3.amazonaws.com/'${REPORT_PREFIX}'/latest/index.html?nocache='"'+Date.now()+'"'" class=|g' ${WORKDIR}/report/app.js
-
-        # Upload a history and the final report (in this particular order to not to have duplicated history in 2 places)
-        time aws s3 mv --recursive --only-show-errors "${WORKDIR}/report/history" "s3://${BUCKET}/${REPORT_PREFIX}/latest/history"
-        time aws s3 mv --recursive --only-show-errors "${WORKDIR}/report" "s3://${BUCKET}/${REPORT_PREFIX}/${GITHUB_RUN_ID}"
-
-        REPORT_URL=https://${BUCKET}.s3.amazonaws.com/${REPORT_PREFIX}/${GITHUB_RUN_ID}/index.html
-
-        # Generate redirect
-        cat <<EOF > ${WORKDIR}/index.html
-          <!DOCTYPE html>
-
-          <meta charset="utf-8">
-          <title>Redirecting to ${REPORT_URL}</title>
-          <meta http-equiv="refresh" content="0; URL=${REPORT_URL}">
-        EOF
-        time aws s3 cp --only-show-errors ${WORKDIR}/index.html "s3://${BUCKET}/${REPORT_PREFIX}/latest/index.html"
-
-        echo "report-url=${REPORT_URL}"                                   >> $GITHUB_OUTPUT
-        echo "report-json-url=${REPORT_URL%/index.html}/data/suites.json" >> $GITHUB_OUTPUT
-
-    - name: Release lock
-      if: always()
-      shell: bash -euxo pipefail {0}
-      run: |
-        aws s3 cp --only-show-errors "s3://${BUCKET}/${LOCK_FILE}" ./lock.txt || exit 0
-
-        if [ "$(cat lock.txt)" = "${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}" ]; then
-          aws s3 rm "s3://${BUCKET}/${LOCK_FILE}"
-        fi
-
-    - name: Cleanup
-      if: always()
-      shell: bash -euxo pipefail {0}
-      run: |
-        if [ -d "${WORKDIR}" ]; then
-          rm -rf ${WORKDIR}
-        fi
-
-    - uses: actions/github-script@v6
-      if: always()
-      env:
-        REPORT_URL: ${{ steps.generate-report.outputs.report-url }}
-        COMMIT_SHA: ${{ github.event.pull_request.head.sha || github.sha }}
-      with:
-        script: |
-          const { REPORT_URL, COMMIT_SHA } = process.env
-
-          await github.rest.repos.createCommitStatus({
-            owner: context.repo.owner,
-            repo: context.repo.repo,
-            sha: `${COMMIT_SHA}`,
-            state: 'success',
-            target_url: `${REPORT_URL}`,
-            context: 'Allure report',
-          })

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

@@ -1,72 +0,0 @@
-name: 'Store Allure results'
-description: 'Upload test results to be used by actions/allure-report-generate'
-
-inputs:
-  report-dir:
-    description: 'directory with test results generated by tests'
-    required: true
-  unique-key:
-    description: 'string to distinguish different results in the same run'
-    required: true
-
-runs:
-  using: "composite"
-
-  steps:
-    - name: Set variables
-      shell: bash -euxo pipefail {0}
-      run: |
-        PR_NUMBER=$(jq --raw-output .pull_request.number "$GITHUB_EVENT_PATH" || true)
-        if [ "${PR_NUMBER}" != "null" ]; then
-          BRANCH_OR_PR=pr-${PR_NUMBER}
-        elif [ "${GITHUB_REF_NAME}" = "main" ] || [ "${GITHUB_REF_NAME}" = "release" ]; then
-          # Shortcut for special branches
-          BRANCH_OR_PR=${GITHUB_REF_NAME}
-        else
-          BRANCH_OR_PR=branch-$(printf "${GITHUB_REF_NAME}" | tr -c "[:alnum:]._-" "-")
-        fi
-
-        echo "BRANCH_OR_PR=${BRANCH_OR_PR}" >> $GITHUB_ENV
-        echo "REPORT_DIR=${REPORT_DIR}"     >> $GITHUB_ENV
-      env:
-        REPORT_DIR: ${{ inputs.report-dir }}
-
-    - name: Upload test results
-      shell: bash -euxo pipefail {0}
-      run: |
-        REPORT_PREFIX=reports/${BRANCH_OR_PR}
-        RAW_PREFIX=reports-raw/${BRANCH_OR_PR}/${GITHUB_RUN_ID}
-
-        # Add metadata
-        cat <<EOF > ${REPORT_DIR}/executor.json
-          {
-            "name": "GitHub Actions",
-            "type": "github",
-            "url": "https://${BUCKET}.s3.amazonaws.com/${REPORT_PREFIX}/latest/index.html",
-            "buildOrder": ${GITHUB_RUN_ID},
-            "buildName": "GitHub Actions Run #${GITHUB_RUN_NUMBER}/${GITHUB_RUN_ATTEMPT}",
-            "buildUrl": "${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}/attempts/${GITHUB_RUN_ATTEMPT}",
-            "reportUrl": "https://${BUCKET}.s3.amazonaws.com/${REPORT_PREFIX}/${GITHUB_RUN_ID}/index.html",
-            "reportName": "Allure Report"
-          }
-        EOF
-
-        cat <<EOF > ${REPORT_DIR}/environment.properties
-          COMMIT_SHA=${COMMIT_SHA}
-        EOF
-
-        ARCHIVE="${UNIQUE_KEY}-${GITHUB_RUN_ATTEMPT}-$(date +%s).tar.zst"
-        ZSTD_NBTHREADS=0
-
-        time tar -C ${REPORT_DIR} -cf ${ARCHIVE} --zstd .
-        time aws s3 mv --only-show-errors ${ARCHIVE} "s3://${BUCKET}/${RAW_PREFIX}/${ARCHIVE}"
-      env:
-        UNIQUE_KEY: ${{ inputs.unique-key }}
-        COMMIT_SHA: ${{ github.event.pull_request.head.sha || github.sha }}
-        BUCKET: neon-github-public-dev
-
-    - name: Cleanup
-      if: always()
-      shell: bash -euxo pipefail {0}
-      run: |
-        rm -rf ${REPORT_DIR}

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

@@ -0,0 +1,254 @@
+name: 'Create Allure report'
+description: 'Create and publish Allure report'
+
+inputs:
+  action:
+    desctiption: 'generate or store'
+    required: true
+  build_type:
+    description: '`build_type` from run-python-test-set action'
+    required: true
+  test_selection:
+    description: '`test_selector` from run-python-test-set action'
+    required: false
+outputs:
+  report-url:
+    description: 'Allure report URL'
+    value: ${{ steps.generate-report.outputs.report-url }}
+  report-json-url:
+    description: 'Allure report JSON URL'
+    value: ${{ steps.generate-report.outputs.report-json-url }}
+
+runs:
+  using: "composite"
+
+  steps:
+    # We're using some of env variables quite offen, so let's set them once.
+    #
+    # It would be nice to have them set in common runs.env[0] section, but it doesn't work[1]
+    #
+    # - [0] https://docs.github.com/en/actions/creating-actions/metadata-syntax-for-github-actions#runsenv
+    # - [1] https://github.com/neondatabase/neon/pull/3907#discussion_r1154703456
+    #
+    - name: Set common environment variables
+      shell: bash -euxo pipefail {0}
+      run: |
+        echo "BUILD_TYPE=${BUILD_TYPE}"   >> $GITHUB_ENV
+        echo "BUCKET=${BUCKET}"           >> $GITHUB_ENV
+        echo "TEST_OUTPUT=${TEST_OUTPUT}" >> $GITHUB_ENV
+      env:
+        BUILD_TYPE: ${{ inputs.build_type }}
+        BUCKET: neon-github-public-dev
+        TEST_OUTPUT: /tmp/test_output
+
+    - name: Validate input parameters
+      shell: bash -euxo pipefail {0}
+      run: |
+        if [ "${{ inputs.action }}" != "store" ] && [ "${{ inputs.action }}" != "generate" ]; then
+          echo >&2 "Unknown inputs.action type '${{ inputs.action }}'; allowed 'generate' or 'store' only"
+          exit 1
+        fi
+
+        if [ -z "${{ inputs.test_selection }}" ] && [ "${{ inputs.action }}" == "store" ]; then
+          echo >&2 "inputs.test_selection must be set for 'store' action"
+          exit 2
+        fi
+
+    - name: Calculate variables
+      id: calculate-vars
+      shell: bash -euxo pipefail {0}
+      run: |
+        # TODO: for manually triggered workflows (via workflow_dispatch) we need to have a separate key
+
+        pr_number=$(jq --raw-output .pull_request.number "$GITHUB_EVENT_PATH" || true)
+        if [ "${pr_number}" != "null" ]; then
+          key=pr-${pr_number}
+        elif [ "${GITHUB_REF_NAME}" = "main" ]; then
+          # Shortcut for a special branch
+          key=main
+        elif [ "${GITHUB_REF_NAME}" = "release" ]; then
+          # Shortcut for a special branch
+          key=release
+        else
+          key=branch-$(printf "${GITHUB_REF_NAME}" | tr -c "[:alnum:]._-" "-")
+        fi
+        echo "KEY=${key}" >> $GITHUB_OUTPUT
+
+        # Sanitize test selection to remove `/` and any other special characters
+        # Use printf instead of echo to avoid having `\n` at the end of the string
+        test_selection=$(printf "${{ inputs.test_selection }}" | tr -c "[:alnum:]._-" "-" )
+        echo "TEST_SELECTION=${test_selection}" >> $GITHUB_OUTPUT
+
+    - uses: actions/setup-java@v3
+      if: ${{ inputs.action == 'generate' }}
+      with:
+        distribution: 'temurin'
+        java-version: '17'
+
+    - name: Install Allure
+      if: ${{ inputs.action == 'generate' }}
+      shell: bash -euxo pipefail {0}
+      run: |
+        if ! which allure; then
+          ALLURE_ZIP=allure-${ALLURE_VERSION}.zip
+          wget -q https://github.com/allure-framework/allure2/releases/download/${ALLURE_VERSION}/${ALLURE_ZIP}
+          echo "${ALLURE_ZIP_MD5}  ${ALLURE_ZIP}" | md5sum -c
+          unzip -q ${ALLURE_ZIP}
+          echo "$(pwd)/allure-${ALLURE_VERSION}/bin" >> $GITHUB_PATH
+          rm -f ${ALLURE_ZIP}
+        fi
+      env:
+        ALLURE_VERSION: 2.21.0
+        ALLURE_ZIP_MD5: c8db4dd8e2a7882583d569ed2c82879c
+
+    - name: Upload Allure results
+      if: ${{ inputs.action == 'store' }}
+      env:
+        REPORT_PREFIX: reports/${{ steps.calculate-vars.outputs.KEY }}/${{ inputs.build_type }}
+        RAW_PREFIX: reports-raw/${{ steps.calculate-vars.outputs.KEY }}/${{ inputs.build_type }}
+        TEST_SELECTION: ${{ steps.calculate-vars.outputs.TEST_SELECTION }}
+      shell: bash -euxo pipefail {0}
+      run: |
+        # Add metadata
+        cat <<EOF > $TEST_OUTPUT/allure/results/executor.json
+          {
+            "name": "GitHub Actions",
+            "type": "github",
+            "url": "https://${BUCKET}.s3.amazonaws.com/${REPORT_PREFIX}/latest/index.html",
+            "buildOrder": ${GITHUB_RUN_ID},
+            "buildName": "GitHub Actions Run #${{ github.run_number }}/${GITHUB_RUN_ATTEMPT}",
+            "buildUrl": "${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}/attempts/${GITHUB_RUN_ATTEMPT}",
+            "reportUrl": "https://${BUCKET}.s3.amazonaws.com/${REPORT_PREFIX}/${GITHUB_RUN_ID}/index.html",
+            "reportName": "Allure Report"
+          }
+        EOF
+        cat <<EOF > $TEST_OUTPUT/allure/results/environment.properties
+          TEST_SELECTION=${{ inputs.test_selection }}
+          BUILD_TYPE=${BUILD_TYPE}
+        EOF
+
+        ARCHIVE="${GITHUB_RUN_ID}-${TEST_SELECTION}-${GITHUB_RUN_ATTEMPT}-$(date +%s).tar.zst"
+        ZSTD_NBTHREADS=0
+
+        tar -C ${TEST_OUTPUT}/allure/results -cf ${ARCHIVE} --zstd .
+        aws s3 mv --only-show-errors ${ARCHIVE} "s3://${BUCKET}/${RAW_PREFIX}/${ARCHIVE}"
+
+    # Potentially we could have several running build for the same key (for example for the main branch), so we use improvised lock for this
+    - name: Acquire Allure lock
+      if: ${{ inputs.action == 'generate' }}
+      shell: bash -euxo pipefail {0}
+      env:
+        LOCK_FILE: reports/${{ steps.calculate-vars.outputs.KEY }}/lock.txt
+        TEST_SELECTION: ${{ steps.calculate-vars.outputs.TEST_SELECTION }}
+      run: |
+        LOCK_TIMEOUT=300 # seconds
+
+        for _ in $(seq 1 5); do
+          for i in $(seq 1 ${LOCK_TIMEOUT}); do
+            LOCK_ADDED=$(aws s3api head-object --bucket neon-github-public-dev --key ${LOCK_FILE} | jq --raw-output '.LastModified' || true)
+            # `date --date="..."` is supported only by gnu date (i.e. it doesn't work on BSD/macOS)
+            if [ -z "${LOCK_ADDED}" ] || [ "$(( $(date +%s) - $(date --date="${LOCK_ADDED}" +%s) ))" -gt "${LOCK_TIMEOUT}" ]; then
+              break
+            fi
+            sleep 1
+          done
+          echo "${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}-${TEST_SELECTION}" > lock.txt
+          aws s3 mv --only-show-errors lock.txt "s3://${BUCKET}/${LOCK_FILE}"
+
+          # A double-check that exactly WE have acquired the lock
+          aws s3 cp --only-show-errors "s3://${BUCKET}/${LOCK_FILE}" ./lock.txt
+          if [ "$(cat lock.txt)" = "${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}-${TEST_SELECTION}" ]; then
+            break
+          fi
+        done
+
+    - name: Generate and publish final Allure report
+      if: ${{ inputs.action == 'generate' }}
+      id: generate-report
+      env:
+        REPORT_PREFIX: reports/${{ steps.calculate-vars.outputs.KEY }}/${{ inputs.build_type }}
+        RAW_PREFIX: reports-raw/${{ steps.calculate-vars.outputs.KEY }}/${{ inputs.build_type }}
+      shell: bash -euxo pipefail {0}
+      run: |
+        # Get previously uploaded data for this run
+        ZSTD_NBTHREADS=0
+
+        s3_filepaths=$(aws s3api list-objects-v2 --bucket ${BUCKET} --prefix ${RAW_PREFIX}/${GITHUB_RUN_ID}- | jq --raw-output  '.Contents[].Key')
+        if [ -z "$s3_filepaths" ]; then
+          # There's no previously uploaded data for this run
+          exit 0
+        fi
+        for s3_filepath in ${s3_filepaths}; do
+          aws s3 cp --only-show-errors "s3://${BUCKET}/${s3_filepath}" "${TEST_OUTPUT}/allure/"
+
+          archive=${TEST_OUTPUT}/allure/$(basename $s3_filepath)
+          mkdir -p ${archive%.tar.zst}
+          tar -xf ${archive} -C ${archive%.tar.zst}
+          rm -f ${archive}
+        done
+
+        # Get history trend
+        aws s3 cp --recursive --only-show-errors "s3://${BUCKET}/${REPORT_PREFIX}/latest/history" "${TEST_OUTPUT}/allure/latest/history" || true
+
+        # Generate report
+        allure generate --clean --output $TEST_OUTPUT/allure/report $TEST_OUTPUT/allure/*
+
+        # Replace a logo link with a redirect to the latest version of the report
+        sed -i 's|<a href="." class=|<a href="https://'${BUCKET}'.s3.amazonaws.com/'${REPORT_PREFIX}'/latest/index.html" class=|g' $TEST_OUTPUT/allure/report/app.js
+
+        # Upload a history and the final report (in this particular order to not to have duplicated history in 2 places)
+        aws s3 mv --recursive --only-show-errors "${TEST_OUTPUT}/allure/report/history" "s3://${BUCKET}/${REPORT_PREFIX}/latest/history"
+        aws s3 mv --recursive --only-show-errors "${TEST_OUTPUT}/allure/report" "s3://${BUCKET}/${REPORT_PREFIX}/${GITHUB_RUN_ID}"
+
+        REPORT_URL=https://${BUCKET}.s3.amazonaws.com/${REPORT_PREFIX}/${GITHUB_RUN_ID}/index.html
+
+        # Generate redirect
+        cat <<EOF > ${TEST_OUTPUT}/allure/index.html
+          <!DOCTYPE html>
+
+          <meta charset="utf-8">
+          <title>Redirecting to ${REPORT_URL}</title>
+          <meta http-equiv="refresh" content="0; URL=${REPORT_URL}">
+        EOF
+        aws s3 cp --only-show-errors ${TEST_OUTPUT}/allure/index.html "s3://${BUCKET}/${REPORT_PREFIX}/latest/index.html"
+
+        echo "[Allure Report](${REPORT_URL})" >> ${GITHUB_STEP_SUMMARY}
+        echo "report-url=${REPORT_URL}" >> $GITHUB_OUTPUT
+        echo "report-json-url=${REPORT_URL%/index.html}/data/suites.json" >> $GITHUB_OUTPUT
+
+    - name: Release Allure lock
+      if: ${{ inputs.action == 'generate' && always() }}
+      shell: bash -euxo pipefail {0}
+      env:
+        LOCK_FILE: reports/${{ steps.calculate-vars.outputs.KEY }}/lock.txt
+        TEST_SELECTION: ${{ steps.calculate-vars.outputs.TEST_SELECTION }}
+      run: |
+        aws s3 cp --only-show-errors "s3://${BUCKET}/${LOCK_FILE}" ./lock.txt || exit 0
+
+        if [ "$(cat lock.txt)" = "${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}-${TEST_SELECTION}" ]; then
+          aws s3 rm "s3://${BUCKET}/${LOCK_FILE}"
+        fi
+
+    - name: Cleanup
+      if: always()
+      shell: bash -euxo pipefail {0}
+      run: |
+        rm -rf ${TEST_OUTPUT}/allure
+
+    - uses: actions/github-script@v6
+      if: ${{ inputs.action == 'generate' && always() }}
+      env:
+        REPORT_URL: ${{ steps.generate-report.outputs.report-url }}
+        SHA: ${{ github.event.pull_request.head.sha || github.sha }}
+      with:
+        script: |
+          const { REPORT_URL, BUILD_TYPE, SHA } = process.env
+
+          await github.rest.repos.createCommitStatus({
+            owner: context.repo.owner,
+            repo: context.repo.repo,
+            sha: `${SHA}`,
+            state: 'success',
+            target_url: `${REPORT_URL}`,
+            context: `Allure report / ${BUILD_TYPE}`,
+          })

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

@@ -197,13 +197,14 @@ runs:
       uses: ./.github/actions/upload
       with:
         name: compatibility-snapshot-${{ inputs.build_type }}-pg14-${{ github.run_id }}
-        # Directory is created by test_compatibility.py::test_create_snapshot, keep the path in sync with the test
-        path: /tmp/test_output/compatibility_snapshot_pg14/
+        # The path includes a test name (test_create_snapshot) and directory that the test creates (compatibility_snapshot_pg14), keep the path in sync with the test
+        path: /tmp/test_output/test_create_snapshot/compatibility_snapshot_pg14/
         prefix: latest
 
-    - name: Upload test results
+    - name: Create Allure report
       if: ${{ !cancelled() }}
-      uses: ./.github/actions/allure-report-store
+      uses: ./.github/actions/allure-report
       with:
-        report-dir: /tmp/test_output/allure/results
-        unique-key: ${{ inputs.build_type }}
+        action: store
+        build_type: ${{ inputs.build_type }}
+        test_selection: ${{ inputs.test_selection }}

.github/workflows/benchmarking.yml

@@ -93,7 +93,10 @@ jobs:
 
     - name: Create Allure report
       if: ${{ !cancelled() }}
-      uses: ./.github/actions/allure-report-generate
+      uses: ./.github/actions/allure-report
+      with:
+        action: generate
+        build_type: ${{ env.BUILD_TYPE }}
 
     - name: Post to a Slack channel
       if: ${{ github.event.schedule && failure() }}
@@ -280,7 +283,10 @@ jobs:
 
     - name: Create Allure report
       if: ${{ !cancelled() }}
-      uses: ./.github/actions/allure-report-generate
+      uses: ./.github/actions/allure-report
+      with:
+        action: generate
+        build_type: ${{ env.BUILD_TYPE }}
 
     - name: Post to a Slack channel
       if: ${{ github.event.schedule && failure() }}
@@ -374,7 +380,10 @@ jobs:
 
     - name: Create Allure report
       if: ${{ !cancelled() }}
-      uses: ./.github/actions/allure-report-generate
+      uses: ./.github/actions/allure-report
+      with:
+        action: generate
+        build_type: ${{ env.BUILD_TYPE }}
 
     - name: Post to a Slack channel
       if: ${{ github.event.schedule && failure() }}
@@ -467,7 +476,10 @@ jobs:
 
     - name: Create Allure report
       if: ${{ !cancelled() }}
-      uses: ./.github/actions/allure-report-generate
+      uses: ./.github/actions/allure-report
+      with:
+        action: generate
+        build_type: ${{ env.BUILD_TYPE }}
 
     - name: Post to a Slack channel
       if: ${{ github.event.schedule && failure() }}
@@ -554,13 +566,16 @@ jobs:
 
     - name: Create Allure report
       if: ${{ !cancelled() }}
-      uses: ./.github/actions/allure-report-generate
+      uses: ./.github/actions/allure-report
+      with:
+        action: generate
+        build_type: ${{ env.BUILD_TYPE }}
 
     - name: Post to a Slack channel
       if: ${{ github.event.schedule && failure() }}
       uses: slackapi/slack-github-action@v1
       with:
         channel-id: "C033QLM5P7D" # dev-staging-stream
-        slack-message: "Periodic User example perf testing ${{ matrix.platform }}: ${{ job.status }}\n${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}"
+        slack-message: "Periodic TPC-H perf testing ${{ matrix.platform }}: ${{ job.status }}\n${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}"
       env:
         SLACK_BOT_TOKEN: ${{ secrets.SLACK_BOT_TOKEN }}

.github/workflows/build_and_test.yml

@@ -330,7 +330,6 @@ jobs:
       fail-fast: false
       matrix:
         build_type: [ debug, release ]
-        pg_version: [ v14, v15 ]
     steps:
       - name: Checkout
         uses: actions/checkout@v3
@@ -351,12 +350,11 @@ jobs:
           real_s3_secret_access_key: "${{ secrets.AWS_SECRET_ACCESS_KEY_CI_TESTS_S3 }}"
           rerun_flaky: true
         env:
-          DEFAULT_PG_VERSION: ${{ matrix.pg_version }}
           TEST_RESULT_CONNSTR: ${{ secrets.REGRESS_TEST_RESULT_CONNSTR }}
           CHECK_ONDISK_DATA_COMPATIBILITY: nonempty
 
       - name: Merge and upload coverage data
-        if: matrix.build_type == 'debug' && matrix.pg_version == 'v14'
+        if: matrix.build_type == 'debug'
         uses: ./.github/actions/save-coverage-data
 
   benchmarks:
@@ -401,10 +399,21 @@ jobs:
     steps:
       - uses: actions/checkout@v3
 
-      - name: Create Allure report
+      - name: Create Allure report (debug)
         if: ${{ !cancelled() }}
-        id: create-allure-report
-        uses: ./.github/actions/allure-report-generate
+        id: create-allure-report-debug
+        uses: ./.github/actions/allure-report
+        with:
+          action: generate
+          build_type: debug
+
+      - name: Create Allure report (release)
+        if: ${{ !cancelled() }}
+        id: create-allure-report-release
+        uses: ./.github/actions/allure-report
+        with:
+          action: generate
+          build_type: release
 
       - uses: actions/github-script@v6
         if: >
@@ -414,37 +423,52 @@ jobs:
           # Retry script for 5XX server errors: https://github.com/actions/github-script#retries
           retries: 5
           script: |
-            const report = {
-              reportUrl:     "${{ steps.create-allure-report.outputs.report-url }}",
-              reportJsonUrl: "${{ steps.create-allure-report.outputs.report-json-url }}",
-            }
+            const reports = [{
+              buildType: "debug",
+              reportUrl: "${{ steps.create-allure-report-debug.outputs.report-url }}",
+              jsonUrl:   "${{ steps.create-allure-report-debug.outputs.report-json-url }}",
+            }, {
+              buildType: "release",
+              reportUrl: "${{ steps.create-allure-report-release.outputs.report-url }}",
+              jsonUrl:   "${{ steps.create-allure-report-release.outputs.report-json-url }}",
+            }]
 
             const script = require("./scripts/pr-comment-test-report.js")
             await script({
               github,
               context,
               fetch,
-              report,
+              reports,
             })
 
       - name: Store Allure test stat in the DB
-        if: ${{ !cancelled() && steps.create-allure-report.outputs.report-json-url }}
+        if: >
+          !cancelled() && (
+            steps.create-allure-report-debug.outputs.report-url ||
+            steps.create-allure-report-release.outputs.report-url
+          )
         env:
-          COMMIT_SHA: ${{ github.event.pull_request.head.sha || github.sha }}
-          REPORT_JSON_URL: ${{ steps.create-allure-report.outputs.report-json-url }}
+          SHA: ${{ github.event.pull_request.head.sha || github.sha }}
+          REPORT_JSON_URL_DEBUG: ${{ steps.create-allure-report-debug.outputs.report-json-url }}
+          REPORT_JSON_URL_RELEASE: ${{ steps.create-allure-report-release.outputs.report-json-url }}
           TEST_RESULT_CONNSTR: ${{ secrets.REGRESS_TEST_RESULT_CONNSTR }}
         run: |
           ./scripts/pysync
 
-          curl --fail --output suites.json "${REPORT_JSON_URL}"
-          export BUILD_TYPE=unified
-          export DATABASE_URL="$TEST_RESULT_CONNSTR"
+          for report_url in $REPORT_JSON_URL_DEBUG $REPORT_JSON_URL_RELEASE; do
+            if [ -z "$report_url" ]; then
+              continue
+            fi
 
-          poetry run python3 scripts/ingest_regress_test_result.py \
-            --revision ${COMMIT_SHA} \
-            --reference ${GITHUB_REF} \
-            --build-type ${BUILD_TYPE} \
-            --ingest suites.json
+            if [[ "$report_url" == "$REPORT_JSON_URL_DEBUG" ]]; then
+              BUILD_TYPE=debug
+            else
+              BUILD_TYPE=release
+            fi
+
+            curl --fail --output suites.json "${report_url}"
+            DATABASE_URL="$TEST_RESULT_CONNSTR" poetry run python3 scripts/ingest_regress_test_result.py --revision ${SHA} --reference ${GITHUB_REF} --build-type ${BUILD_TYPE} --ingest suites.json
+          done
 
   coverage-report:
     runs-on: [ self-hosted, gen3, small ]

libs/pageserver_api/src/models.rs

@@ -48,33 +48,13 @@ pub enum TenantState {
 }
 
 impl TenantState {
-    pub fn attachment_status(&self) -> TenantAttachmentStatus {
-        use TenantAttachmentStatus::*;
+    pub fn has_in_progress_downloads(&self) -> bool {
         match self {
-            // The attach procedure writes the marker file before adding the Attaching tenant to the tenants map.
-            // So, technically, we can return Attached here.
-            // However, as soon as Console observes Attached, it will proceed with the Postgres-level health check.
-            // But, our attach task might still be fetching the remote timelines, etc.
-            // So, return `Maybe` while Attaching, making Console wait for the attach task to finish.
-            Self::Attaching => Maybe,
-            // tenant mgr startup distinguishes attaching from loading via marker file.
-            // If it's loading, there is no attach marker file, i.e., attach had finished in the past.
-            Self::Loading => Attached,
-            // We only reach Active after successful load / attach.
-            // So, call atttachment status Attached.
-            Self::Active => Attached,
-            // If the (initial or resumed) attach procedure fails, the tenant becomes Broken.
-            // However, it also becomes Broken if the regular load fails.
-            // We would need a separate TenantState variant to distinguish these cases.
-            // However, there's no practical difference from Console's perspective.
-            // It will run a Postgres-level health check as soon as it observes Attached.
-            // That will fail on Broken tenants.
-            // Console can then rollback the attach, or, wait for operator to fix the Broken tenant.
-            Self::Broken { .. } => Attached,
-            // Why is Stopping a Maybe case? Because, during pageserver shutdown,
-            // we set the Stopping state irrespective of whether the tenant
-            // has finished attaching or not.
-            Self::Stopping => Maybe,
+            Self::Loading => true,
+            Self::Attaching => true,
+            Self::Active => false,
+            Self::Stopping => false,
+            Self::Broken { .. } => false,
         }
     }
 
@@ -229,25 +209,16 @@ impl TenantConfigRequest {
     }
 }
 
-/// See [`TenantState::attachment_status`] and the OpenAPI docs for context.
-#[derive(Serialize, Deserialize, Clone)]
-#[serde(rename_all = "snake_case")]
-pub enum TenantAttachmentStatus {
-    Maybe,
-    Attached,
-}
-
 #[serde_as]
 #[derive(Serialize, Deserialize, Clone)]
 pub struct TenantInfo {
     #[serde_as(as = "DisplayFromStr")]
     pub id: TenantId,
-    // NB: intentionally not part of OpenAPI, we don't want to commit to a specific set of TenantState's
     pub state: TenantState,
     /// Sum of the size of all layer files.
     /// If a layer is present in both local FS and S3, it counts only once.
     pub current_physical_size: Option<u64>, // physical size is only included in `tenant_status` endpoint
-    pub attachment_status: TenantAttachmentStatus,
+    pub has_in_progress_downloads: Option<bool>,
 }
 
 /// This represents the output of the "timeline_detail" and "timeline_list" API calls.
@@ -720,7 +691,7 @@ mod tests {
             id: TenantId::generate(),
             state: TenantState::Active,
             current_physical_size: Some(42),
-            attachment_status: TenantAttachmentStatus::Attached,
+            has_in_progress_downloads: Some(false),
         };
         let expected_active = json!({
             "id": original_active.id.to_string(),
@@ -728,7 +699,7 @@ mod tests {
                 "slug": "Active",
             },
             "current_physical_size": 42,
-            "attachment_status": "attached",
+            "has_in_progress_downloads": false,
         });
 
         let original_broken = TenantInfo {
@@ -738,7 +709,7 @@ mod tests {
                 backtrace: "backtrace info".into(),
             },
             current_physical_size: Some(42),
-            attachment_status: TenantAttachmentStatus::Attached,
+            has_in_progress_downloads: Some(false),
         };
         let expected_broken = json!({
             "id": original_broken.id.to_string(),
@@ -750,7 +721,7 @@ mod tests {
                 }
             },
             "current_physical_size": 42,
-            "attachment_status": "attached",
+            "has_in_progress_downloads": false,
         });
 
         assert_eq!(

libs/postgres_ffi/src/pg_constants.rs

@@ -146,10 +146,6 @@ pub const XLH_UPDATE_OLD_ALL_VISIBLE_CLEARED: u8 = (1 << 0) as u8;
 pub const XLH_UPDATE_NEW_ALL_VISIBLE_CLEARED: u8 = (1 << 1) as u8;
 pub const XLH_DELETE_ALL_VISIBLE_CLEARED: u8 = (1 << 0) as u8;
 
-// From replication/message.h
-pub const XLOG_LOGICAL_MESSAGE: u8 = 0x00;
-
-// From rmgrlist.h
 pub const RM_XLOG_ID: u8 = 0;
 pub const RM_XACT_ID: u8 = 1;
 pub const RM_SMGR_ID: u8 = 2;
@@ -161,7 +157,6 @@ pub const RM_RELMAP_ID: u8 = 7;
 pub const RM_STANDBY_ID: u8 = 8;
 pub const RM_HEAP2_ID: u8 = 9;
 pub const RM_HEAP_ID: u8 = 10;
-pub const RM_LOGICALMSG_ID: u8 = 21;
 
 // from xlogreader.h
 pub const XLR_INFO_MASK: u8 = 0x0F;

libs/utils/scripts/restore_from_wal.sh

@@ -1,21 +1,21 @@
 #!/bin/bash
-
-set -euxo pipefail
-
 PG_BIN=$1
 WAL_PATH=$2
 DATA_DIR=$3
 PORT=$4
-SYSID=$(od -A n -j 24 -N 8 -t d8 "$WAL_PATH"/000000010000000000000002* | cut -c 3-)
-rm -fr "$DATA_DIR"
-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
-REDO_POS=0x$("$PG_BIN"/pg_controldata -D "$DATA_DIR" | grep -F "REDO location"| cut -c 42-)
+SYSID=`od -A n -j 24 -N 8 -t d8 $WAL_PATH/000000010000000000000002* | cut -c 3-`
+rm -fr $DATA_DIR
+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
+REDO_POS=0x`$PG_BIN/pg_controldata -D $DATA_DIR | fgrep "REDO location"| cut -c 42-`
 declare -i WAL_SIZE=$REDO_POS+114
-"$PG_BIN"/pg_ctl -D "$DATA_DIR" -l logfile start
-"$PG_BIN"/pg_ctl -D "$DATA_DIR" -l logfile stop -m immediate
-cp "$DATA_DIR"/pg_wal/000000010000000000000001 .
-cp "$WAL_PATH"/* "$DATA_DIR"/pg_wal/
-for partial in "$DATA_DIR"/pg_wal/*.partial ; do mv "$partial" "${partial%.partial}" ; done
-dd if=000000010000000000000001 of="$DATA_DIR"/pg_wal/000000010000000000000001 bs=$WAL_SIZE count=1 conv=notrunc
+$PG_BIN/pg_ctl -D $DATA_DIR -l logfile start
+$PG_BIN/pg_ctl -D $DATA_DIR -l logfile stop -m immediate
+cp $DATA_DIR/pg_wal/000000010000000000000001 .
+cp $WAL_PATH/* $DATA_DIR/pg_wal/
+if [ -f $DATA_DIR/pg_wal/*.partial ]
+then
+	(cd $DATA_DIR/pg_wal ; for partial in \*.partial ; do mv $partial `basename $partial .partial` ; done)
+fi
+dd if=000000010000000000000001 of=$DATA_DIR/pg_wal/000000010000000000000001 bs=$WAL_SIZE count=1 conv=notrunc
 rm -f 000000010000000000000001

libs/utils/scripts/restore_from_wal_archive.sh

@@ -0,0 +1,20 @@
+PG_BIN=$1
+WAL_PATH=$2
+DATA_DIR=$3
+PORT=$4
+SYSID=`od -A n -j 24 -N 8 -t d8 $WAL_PATH/000000010000000000000002* | cut -c 3-`
+rm -fr $DATA_DIR /tmp/pg_wals
+mkdir /tmp/pg_wals
+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
+REDO_POS=0x`$PG_BIN/pg_controldata -D $DATA_DIR | fgrep "REDO location"| cut -c 42-`
+declare -i WAL_SIZE=$REDO_POS+114
+cp $WAL_PATH/* /tmp/pg_wals
+if [ -f $DATA_DIR/pg_wal/*.partial ]
+then
+	(cd /tmp/pg_wals ; for partial in \*.partial ; do  mv $partial `basename $partial .partial` ; done)
+fi
+dd if=$DATA_DIR/pg_wal/000000010000000000000001 of=/tmp/pg_wals/000000010000000000000001 bs=$WAL_SIZE count=1 conv=notrunc
+echo > $DATA_DIR/recovery.signal
+rm -f $DATA_DIR/pg_wal/*
+echo "restore_command = 'cp /tmp/pg_wals/%f %p'" >> $DATA_DIR/postgresql.conf

libs/utils/src/lib.rs

@@ -60,43 +60,28 @@ pub mod tracing_span_assert;
 
 pub mod rate_limit;
 
-mod failpoint_macro_helpers {
+/// Primitive for coalescing operations into a single task which will not be cancelled by for
+/// example external http client closing the connection.
+pub mod shared_retryable;
 
-    /// use with fail::cfg("$name", "return(2000)")
-    ///
-    /// The effect is similar to a "sleep(2000)" action, i.e. we sleep for the
-    /// specified time (in milliseconds). The main difference is that we use async
-    /// tokio sleep function. Another difference is that we print lines to the log,
-    /// which can be useful in tests to check that the failpoint was hit.
-    #[macro_export]
-    macro_rules! failpoint_sleep_millis_async {
-        ($name:literal) => {{
-            // If the failpoint is used with a "return" action, set should_sleep to the
-            // returned value (as string). Otherwise it's set to None.
-            let should_sleep = (|| {
-                ::fail::fail_point!($name, |x| x);
-                ::std::option::Option::None
-            })();
-
-            // Sleep if the action was a returned value
-            if let ::std::option::Option::Some(duration_str) = should_sleep {
-                $crate::failpoint_sleep_helper($name, duration_str).await
-            }
-        }};
-    }
-
-    // Helper function used by the macro. (A function has nicer scoping so we
-    // don't need to decorate everything with "::")
-    pub async fn failpoint_sleep_helper(name: &'static str, duration_str: String) {
-        let millis = duration_str.parse::<u64>().unwrap();
-        let d = std::time::Duration::from_millis(millis);
-
-        tracing::info!("failpoint {:?}: sleeping for {:?}", name, d);
-        tokio::time::sleep(d).await;
-        tracing::info!("failpoint {:?}: sleep done", name);
-    }
+/// use with fail::cfg("$name", "return(2000)")
+#[macro_export]
+macro_rules! failpoint_sleep_millis_async {
+    ($name:literal) => {{
+        let should_sleep: Option<std::time::Duration> = (|| {
+            fail::fail_point!($name, |v: Option<_>| {
+                let millis = v.unwrap().parse::<u64>().unwrap();
+                Some(Duration::from_millis(millis))
+            });
+            None
+        })();
+        if let Some(d) = should_sleep {
+            tracing::info!("failpoint {:?}: sleeping for {:?}", $name, d);
+            tokio::time::sleep(d).await;
+            tracing::info!("failpoint {:?}: sleep done", $name);
+        }
+    }};
 }
-pub use failpoint_macro_helpers::failpoint_sleep_helper;
 
 /// This is a shortcut to embed git sha into binaries and avoid copying the same build script to all packages
 ///

libs/utils/src/shared_retryable.rs

@@ -0,0 +1,659 @@
+use std::future::Future;
+use std::sync::Arc;
+
+/// Container using which many request handlers can come together and join a single task to
+/// completion instead of racing each other and their own cancellation.
+///
+/// In a picture:
+///
+/// ```text
+/// SharedRetryable::try_restart         Spawned task completes with only one concurrent attempt
+///                             \       /
+///      request handler 1 ---->|--X
+///      request handler 2 ---->|-------|
+///      request handler 3 ---->|-------|
+///                             |       |
+///                             v       |
+///       one spawned task      \------>/
+///
+/// (X = cancelled during await)
+/// ```
+///
+/// Implementation is cancel safe. Implementation and internal structure are hurt by the inability
+/// to just spawn the task, but this is needed for pageserver usage.
+///
+/// Implementation exposes a fully decomposed [`SharedRetryable::try_restart`] which requires the
+/// caller to do the spawning before awaiting for the result. If the caller is dropped while this
+/// happens, a new attempt will be required, and all concurrent awaiters will see a
+/// [`RetriedTaskPanicked`] error.
+///
+/// There is another "family of APIs" [`SharedRetryable::attempt_spawn`] for infallible futures. It is
+/// just provided for completeness, and it does not have a fully decomposed version like
+/// `try_restart`.
+///
+/// For `try_restart_*` family of APIs, there is a concept of two leveled results. The inner level
+/// is returned by the executed future. It needs to be `Clone`. Most errors are not `Clone`, so
+/// implementation advice is to log the happened error, and not propagate more than a label as the
+/// "inner error" which will be used to build an outer error. The outer error will also have to be
+/// convertable from [`RetriedTaskPanicked`] to absorb that case as well.
+///
+/// ## Example
+///
+/// A shared service value completes the infallible work once, even if called concurrently by
+/// multiple cancellable tasks.
+///
+/// Example moved as a test `service_example`.
+#[derive(Clone)]
+pub struct SharedRetryable<V> {
+    inner: Arc<tokio::sync::Mutex<Option<MaybeDone<V>>>>,
+}
+
+impl<V> Default for SharedRetryable<V> {
+    fn default() -> Self {
+        Self {
+            inner: Arc::new(tokio::sync::Mutex::new(None)),
+        }
+    }
+}
+
+/// Determine if an error is transient or permanent.
+pub trait Retryable {
+    fn is_permanent(&self) -> bool {
+        true
+    }
+}
+
+pub trait MakeFuture {
+    type Future: Future<Output = Self::Output> + Send + 'static;
+    type Output: Send + 'static;
+
+    fn make_future(self) -> Self::Future;
+}
+
+impl<Fun, Fut, R> MakeFuture for Fun
+where
+    Fun: FnOnce() -> Fut,
+    Fut: Future<Output = R> + Send + 'static,
+    R: Send + 'static,
+{
+    type Future = Fut;
+    type Output = R;
+
+    fn make_future(self) -> Self::Future {
+        self()
+    }
+}
+
+/// Retried task panicked, was cancelled, or never spawned (see [`SharedRetryable::try_restart`]).
+#[derive(Debug, PartialEq, Eq)]
+pub struct RetriedTaskPanicked;
+
+impl<T, E1> SharedRetryable<Result<T, E1>>
+where
+    T: Clone + std::fmt::Debug + Send + 'static,
+    E1: Retryable + Clone + std::fmt::Debug + Send + 'static,
+{
+    /// Restart a previously failed operation unless it already completed with a terminal result.
+    ///
+    /// Many futures can call this function and and get the terminal result from an earlier attempt
+    /// or start a new attempt, or join an existing one.
+    ///
+    /// Compared to `Self::try_restart`, this method also spawns the future to run, which would
+    /// otherwise have to be done manually.
+    #[cfg(test)]
+    pub async fn try_restart_spawn<E2>(
+        &self,
+        retry_with: impl MakeFuture<Output = Result<T, E1>>,
+    ) -> Result<T, E2>
+    where
+        E2: From<E1> + From<RetriedTaskPanicked> + Send + 'static,
+    {
+        let (recv, maybe_fut) = self.try_restart(retry_with).await;
+
+        if let Some(fut) = maybe_fut {
+            // top level function, we must spawn, pageserver cannot use this
+            tokio::spawn(fut);
+        }
+
+        recv.await
+    }
+
+    /// Restart a previously failed operation unless it already completed with a terminal result.
+    ///
+    /// Many futures can call this function and get the terminal result from an earlier attempt or
+    /// start a new attempt, or join an existing one.
+    ///
+    /// If a task calling this method is cancelled, at worst, a new attempt which is immediatedly
+    /// deemed as having panicked will happen, but without a panic ever happening.
+    ///
+    /// Returns one future for waiting for the result and possibly another which needs to be
+    /// spawned when `Some`. Spawning has to happen before waiting is started, otherwise the first
+    /// future will never make progress.
+    ///
+    /// This complication exists because on pageserver we cannot use `tokio::spawn` directly
+    /// at this time.
+    pub async fn try_restart<E2>(
+        &self,
+        retry_with: impl MakeFuture<Output = Result<T, E1>>,
+    ) -> (
+        impl Future<Output = Result<T, E2>> + Send + 'static,
+        Option<impl Future<Output = ()> + Send + 'static>,
+    )
+    where
+        E2: From<E1> + From<RetriedTaskPanicked> + Send + 'static,
+    {
+        use futures::future::Either;
+
+        match self.decide_to_retry_or_join(retry_with).await {
+            Ok(terminal) => (Either::Left(async move { terminal }), None),
+            Err((rx, maybe_fut)) => {
+                let recv = Self::make_oneshot_alike_receiver(rx);
+
+                (Either::Right(recv), maybe_fut)
+            }
+        }
+    }
+
+    /// Returns a Ok if the previous attempt had resulted in a terminal result. Err is returned
+    /// when an attempt can be joined and possibly needs to be spawned.
+    async fn decide_to_retry_or_join<E2>(
+        &self,
+        retry_with: impl MakeFuture<Output = Result<T, E1>>,
+    ) -> Result<
+        Result<T, E2>,
+        (
+            tokio::sync::broadcast::Receiver<Result<T, E1>>,
+            Option<impl Future<Output = ()> + Send + 'static>,
+        ),
+    >
+    where
+        E2: From<E1> + From<RetriedTaskPanicked>,
+    {
+        let mut g = self.inner.lock().await;
+
+        let maybe_rx = match g.as_ref() {
+            Some(MaybeDone::Done(Ok(t))) => return Ok(Ok(t.to_owned())),
+            Some(MaybeDone::Done(Err(e))) if e.is_permanent() => {
+                return Ok(Err(E2::from(e.to_owned())))
+            }
+            Some(MaybeDone::Pending(weak)) => {
+                // failure to upgrade can mean only one thing: there was an unexpected
+                // panic which we consider as a transient retryable error.
+                weak.upgrade()
+            }
+            Some(MaybeDone::Done(Err(_retryable))) => None,
+            None => None,
+        };
+
+        let (strong, maybe_fut) = match maybe_rx {
+            Some(strong) => (strong, None),
+            None => {
+                // new attempt
+                // panic safety: invoke the factory before configuring the pending value
+                let fut = retry_with.make_future();
+
+                let (strong, fut) = self.make_run_and_complete(fut, &mut g);
+                (strong, Some(fut))
+            }
+        };
+
+        // important: the Arc<Receiver> is not held after unlocking
+        // important: we resubscribe before lock is released to be sure to get a message which
+        // is sent once receiver is dropped
+        let rx = strong.resubscribe();
+        drop(strong);
+        Err((rx, maybe_fut))
+    }
+
+    /// Configure a new attempt, but leave spawning it to the caller.
+    ///
+    /// Returns an `Arc<Receiver<V>>` which is valid until the attempt completes, and the future
+    /// which will need to run to completion outside the lifecycle of the caller.
+    fn make_run_and_complete(
+        &self,
+        fut: impl Future<Output = Result<T, E1>> + Send + 'static,
+        g: &mut tokio::sync::MutexGuard<'_, Option<MaybeDone<Result<T, E1>>>>,
+    ) -> (
+        Arc<tokio::sync::broadcast::Receiver<Result<T, E1>>>,
+        impl Future<Output = ()> + Send + 'static,
+    ) {
+        #[cfg(debug_assertions)]
+        match &**g {
+            Some(MaybeDone::Pending(weak)) => {
+                assert!(
+                    weak.upgrade().is_none(),
+                    "when starting a restart, should no longer have an upgradeable channel"
+                );
+            }
+            Some(MaybeDone::Done(Err(err))) => {
+                assert!(
+                    !err.is_permanent(),
+                    "when restarting, the err must be transient"
+                );
+            }
+            Some(MaybeDone::Done(Ok(_))) => {
+                panic!("unexpected restart after a completion on MaybeDone");
+            }
+            None => {}
+        }
+
+        self.make_run_and_complete_any(fut, g)
+    }
+
+    /// Oneshot alike as in it's a future which will be consumed by an `await`.
+    ///
+    /// Otherwise the caller might think it's beneficial or reasonable to poll the channel multiple
+    /// times.
+    async fn make_oneshot_alike_receiver<E2>(
+        mut rx: tokio::sync::broadcast::Receiver<Result<T, E1>>,
+    ) -> Result<T, E2>
+    where
+        E2: From<E1> + From<RetriedTaskPanicked>,
+    {
+        use tokio::sync::broadcast::error::RecvError;
+
+        match rx.recv().await {
+            Ok(Ok(t)) => Ok(t),
+            Ok(Err(e)) => Err(E2::from(e)),
+            Err(RecvError::Closed | RecvError::Lagged(_)) => {
+                // lagged doesn't mean anything with 1 send, but whatever, handle it the same
+                // this case should only ever happen if a panick happened in the `fut`.
+                Err(E2::from(RetriedTaskPanicked))
+            }
+        }
+    }
+}
+
+impl<V> SharedRetryable<V>
+where
+    V: std::fmt::Debug + Clone + Send + 'static,
+{
+    /// Attempt to run once a spawned future to completion.
+    ///
+    /// Any previous attempt which panicked will be retried, but the `RetriedTaskPanicked` will be
+    /// returned when the most recent attempt panicked.
+    #[cfg(test)]
+    pub async fn attempt_spawn(
+        &self,
+        attempt_with: impl MakeFuture<Output = V>,
+    ) -> Result<V, RetriedTaskPanicked> {
+        let (rx, maybe_fut) = {
+            let mut g = self.inner.lock().await;
+
+            let maybe_rx = match g.as_ref() {
+                Some(MaybeDone::Done(v)) => return Ok(v.to_owned()),
+                Some(MaybeDone::Pending(weak)) => {
+                    // see comment in try_restart
+                    weak.upgrade()
+                }
+                None => None,
+            };
+
+            let (strong, maybe_fut) = match maybe_rx {
+                Some(strong) => (strong, None),
+                None => {
+                    let fut = attempt_with.make_future();
+
+                    let (strong, fut) = self.make_run_and_complete_any(fut, &mut g);
+                    (strong, Some(fut))
+                }
+            };
+
+            // see decide_to_retry_or_join for important notes
+            let rx = strong.resubscribe();
+            drop(strong);
+            (rx, maybe_fut)
+        };
+
+        if let Some(fut) = maybe_fut {
+            // this is a top level function, need to spawn directly
+            // from pageserver one wouldn't use this but more piecewise functions
+            tokio::spawn(fut);
+        }
+
+        let recv = Self::make_oneshot_alike_receiver_any(rx);
+
+        recv.await
+    }
+
+    /// Configure a new attempt, but leave spawning it to the caller.
+    ///
+    /// Forgetting the returned future is outside of scope of any correctness guarantees; all of
+    /// the waiters will then be deadlocked, and the MaybeDone will forever be pending. Dropping
+    /// and not running the future will then require a new attempt.
+    ///
+    /// Also returns an `Arc<Receiver<V>>` which is valid until the attempt completes.
+    fn make_run_and_complete_any(
+        &self,
+        fut: impl Future<Output = V> + Send + 'static,
+        g: &mut tokio::sync::MutexGuard<'_, Option<MaybeDone<V>>>,
+    ) -> (
+        Arc<tokio::sync::broadcast::Receiver<V>>,
+        impl Future<Output = ()> + Send + 'static,
+    ) {
+        let (tx, rx) = tokio::sync::broadcast::channel(1);
+        let strong = Arc::new(rx);
+
+        **g = Some(MaybeDone::Pending(Arc::downgrade(&strong)));
+
+        let retry = {
+            let strong = strong.clone();
+            self.clone().run_and_complete(fut, tx, strong)
+        };
+
+        #[cfg(debug_assertions)]
+        match &**g {
+            Some(MaybeDone::Pending(weak)) => {
+                let rx = weak.upgrade().expect("holding the weak and strong locally");
+                assert!(Arc::ptr_eq(&strong, &rx));
+            }
+            _ => unreachable!("MaybeDone::pending must be set after spawn_and_run_complete_any"),
+        }
+
+        (strong, retry)
+    }
+
+    /// Run the actual attempt, and communicate the response via both:
+    /// - setting the `MaybeDone::Done`
+    /// - the broadcast channel
+    async fn run_and_complete(
+        self,
+        fut: impl Future<Output = V>,
+        tx: tokio::sync::broadcast::Sender<V>,
+        strong: Arc<tokio::sync::broadcast::Receiver<V>>,
+    ) {
+        let res = fut.await;
+
+        {
+            let mut g = self.inner.lock().await;
+            MaybeDone::complete(&mut *g, &strong, res.clone());
+
+            // make the weak un-upgradeable by dropping the final alive
+            // reference to it. it is final Arc because the Arc never escapes
+            // the critical section in `decide_to_retry_or_join` or `attempt_spawn`.
+            Arc::try_unwrap(strong).expect("expected this to be the only Arc<Receiver<V>>");
+        }
+
+        // now no one can get the Pending(weak) value to upgrade and they only see
+        // the Done(res).
+        //
+        // send the result value to listeners, if any
+        drop(tx.send(res));
+    }
+
+    #[cfg(test)]
+    async fn make_oneshot_alike_receiver_any(
+        mut rx: tokio::sync::broadcast::Receiver<V>,
+    ) -> Result<V, RetriedTaskPanicked> {
+        use tokio::sync::broadcast::error::RecvError;
+
+        match rx.recv().await {
+            Ok(t) => Ok(t),
+            Err(RecvError::Closed | RecvError::Lagged(_)) => {
+                // lagged doesn't mean anything with 1 send, but whatever, handle it the same
+                // this case should only ever happen if a panick happened in the `fut`.
+                Err(RetriedTaskPanicked)
+            }
+        }
+    }
+}
+
+/// MaybeDone handles synchronization for multiple requests and the single actual task.
+///
+/// If request handlers witness `Pending` which they are able to upgrade, they are guaranteed a
+/// useful `recv().await`, where useful means "value" or "disconnect" arrives. If upgrade fails,
+/// this means that "disconnect" has happened in the past.
+///
+/// On successful execution the one executing task will set this to `Done` variant, with the actual
+/// resulting value.
+#[derive(Debug)]
+pub enum MaybeDone<V> {
+    Pending(std::sync::Weak<tokio::sync::broadcast::Receiver<V>>),
+    Done(V),
+}
+
+impl<V: std::fmt::Debug> MaybeDone<V> {
+    fn complete(
+        this: &mut Option<MaybeDone<V>>,
+        _strong: &Arc<tokio::sync::broadcast::Receiver<V>>,
+        outcome: V,
+    ) {
+        #[cfg(debug_assertions)]
+        match this {
+            Some(MaybeDone::Pending(weak)) => {
+                let same = weak
+                    .upgrade()
+                    // we don't yet have Receiver::same_channel
+                    .map(|rx| Arc::ptr_eq(_strong, &rx))
+                    .unwrap_or(false);
+                assert!(same, "different channel had been replaced or dropped");
+            }
+            other => panic!("unexpected MaybeDone: {other:?}"),
+        }
+
+        *this = Some(MaybeDone::Done(outcome));
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{RetriedTaskPanicked, Retryable, SharedRetryable};
+    use std::sync::Arc;
+
+    #[derive(Debug)]
+    enum OuterError {
+        AttemptPanicked,
+        Unlucky,
+    }
+
+    #[derive(Clone, Debug)]
+    enum InnerError {
+        Unlucky,
+    }
+
+    impl Retryable for InnerError {
+        fn is_permanent(&self) -> bool {
+            false
+        }
+    }
+
+    impl From<InnerError> for OuterError {
+        fn from(_: InnerError) -> Self {
+            OuterError::Unlucky
+        }
+    }
+
+    impl From<RetriedTaskPanicked> for OuterError {
+        fn from(_: RetriedTaskPanicked) -> Self {
+            OuterError::AttemptPanicked
+        }
+    }
+
+    #[tokio::test]
+    async fn restartable_until_permanent() {
+        let shr = SharedRetryable::<Result<u8, InnerError>>::default();
+
+        let res = shr
+            .try_restart_spawn(|| async move { panic!("really unlucky") })
+            .await;
+
+        assert!(matches!(res, Err(OuterError::AttemptPanicked)));
+
+        let res = shr
+            .try_restart_spawn(|| async move { Err(InnerError::Unlucky) })
+            .await;
+
+        assert!(matches!(res, Err(OuterError::Unlucky)));
+
+        let res = shr.try_restart_spawn(|| async move { Ok(42) }).await;
+
+        assert!(matches!(res, Ok::<u8, OuterError>(42)));
+
+        let res = shr
+            .try_restart_spawn(|| async move { panic!("rerun should clone Ok(42)") })
+            .await;
+
+        assert!(matches!(res, Ok::<u8, OuterError>(42)));
+    }
+
+    /// Demonstration of the SharedRetryable::attempt
+    #[tokio::test]
+    async fn attemptable_until_no_panic() {
+        let shr = SharedRetryable::<u8>::default();
+
+        let res = shr
+            .attempt_spawn(|| async move { panic!("should not interfere") })
+            .await;
+
+        assert!(matches!(res, Err(RetriedTaskPanicked)), "{res:?}");
+
+        let res = shr.attempt_spawn(|| async move { 42 }).await;
+
+        assert_eq!(res, Ok(42));
+
+        let res = shr
+            .attempt_spawn(|| async move { panic!("should not be called") })
+            .await;
+
+        assert_eq!(res, Ok(42));
+    }
+
+    #[tokio::test]
+    async fn cancelling_spawner_is_fine() {
+        let shr = SharedRetryable::<Result<u8, InnerError>>::default();
+
+        let (recv1, maybe_fut) = shr
+            .try_restart(|| async move { panic!("should not have been called") })
+            .await;
+        let should_be_spawned = maybe_fut.unwrap();
+
+        let (recv2, maybe_fut) = shr
+            .try_restart(|| async move {
+                panic!("should never be called because waiting on should_be_spawned")
+            })
+            .await;
+        assert!(
+            matches!(maybe_fut, None),
+            "only the first one should had created the future"
+        );
+
+        let mut recv1 = std::pin::pin!(recv1);
+        let mut recv2 = std::pin::pin!(recv2);
+
+        tokio::select! {
+            _ = tokio::time::sleep(std::time::Duration::from_millis(100)) => {},
+            _ = &mut recv1 => unreachable!("should not have completed because should_be_spawned not spawned"),
+            _ = &mut recv2 => unreachable!("should not have completed because should_be_spawned not spawned"),
+        }
+
+        drop(should_be_spawned);
+
+        let res = recv1.await;
+        assert!(matches!(res, Err(OuterError::AttemptPanicked)), "{res:?}");
+
+        let res = recv2.await;
+        assert!(matches!(res, Err(OuterError::AttemptPanicked)), "{res:?}");
+
+        // but we can still reach a terminal state if the api is not misused or the
+        // should_be_spawned winner is not cancelled
+
+        let recv1 = shr.try_restart_spawn::<OuterError>(|| async move { Ok(42) });
+        let recv2 = shr.try_restart_spawn::<OuterError>(|| async move { Ok(43) });
+
+        assert_eq!(recv1.await.unwrap(), 42);
+        assert_eq!(recv2.await.unwrap(), 42, "43 should never be returned");
+    }
+
+    #[tokio::test]
+    async fn service_example() {
+        #[derive(Debug, Clone, Copy)]
+        enum OneLevelError {
+            TaskPanicked,
+        }
+
+        impl Retryable for OneLevelError {
+            fn is_permanent(&self) -> bool {
+                // for a single level errors, this wording is weird
+                !matches!(self, OneLevelError::TaskPanicked)
+            }
+        }
+
+        impl From<RetriedTaskPanicked> for OneLevelError {
+            fn from(_: RetriedTaskPanicked) -> Self {
+                OneLevelError::TaskPanicked
+            }
+        }
+
+        #[derive(Clone, Default)]
+        struct Service(SharedRetryable<Result<u8, OneLevelError>>);
+
+        impl Service {
+            async fn work(
+                &self,
+                completions: Arc<std::sync::atomic::AtomicUsize>,
+            ) -> Result<u8, OneLevelError> {
+                self.0
+                    .try_restart_spawn(|| async move {
+                        // give time to cancel some of the tasks
+                        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+                        completions.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+                        Self::work_once().await
+                    })
+                    .await
+            }
+
+            async fn work_once() -> Result<u8, OneLevelError> {
+                Ok(42)
+            }
+        }
+
+        let svc = Service::default();
+
+        let mut js = tokio::task::JoinSet::new();
+
+        let barrier = Arc::new(tokio::sync::Barrier::new(10 + 1));
+        let completions = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+
+        let handles = (0..10)
+            .map(|_| {
+                js.spawn({
+                    let svc = svc.clone();
+                    let barrier = barrier.clone();
+                    let completions = completions.clone();
+                    async move {
+                        // make sure all tasks are ready to start at the same time
+                        barrier.wait().await;
+                        // after successfully starting the work, any of the futures could get cancelled
+                        svc.work(completions).await
+                    }
+                })
+            })
+            .collect::<Vec<_>>();
+
+        barrier.wait().await;
+
+        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+
+        handles[5].abort();
+
+        let mut cancellations = 0;
+
+        while let Some(res) = js.join_next().await {
+            // all complete with the same result
+            match res {
+                Ok(res) => assert_eq!(res.unwrap(), 42),
+                Err(je) => {
+                    // except for the one task we cancelled; it's cancelling
+                    // does not interfere with the result
+                    assert!(je.is_cancelled());
+                    cancellations += 1;
+                    assert_eq!(cancellations, 1, "only 6th task was aborted");
+                    // however we cannot assert that everytime we get to cancel the 6th task
+                }
+            }
+        }
+
+        // there will be at most one terminal completion
+        assert_eq!(completions.load(std::sync::atomic::Ordering::Relaxed), 1);
+    }
+}

pageserver/src/consumption_metrics.rs

@@ -5,7 +5,7 @@
 //!
 use crate::context::{DownloadBehavior, RequestContext};
 use crate::task_mgr::{self, TaskKind, BACKGROUND_RUNTIME};
-use crate::tenant::{mgr, LogicalSizeCalculationCause};
+use crate::tenant::mgr;
 use anyhow;
 use chrono::Utc;
 use consumption_metrics::{idempotency_key, Event, EventChunk, EventType, CHUNK_SIZE};
@@ -164,8 +164,7 @@ pub async fn collect_metrics_iteration(
                     timeline_written_size,
                 ));
 
-                let span = info_span!("collect_metrics_iteration", tenant_id = %timeline.tenant_id, timeline_id = %timeline.timeline_id);
-                match span.in_scope(|| timeline.get_current_logical_size(ctx)) {
+                match timeline.get_current_logical_size(ctx) {
                     // Only send timeline logical size when it is fully calculated.
                     Ok((size, is_exact)) if is_exact => {
                         current_metrics.push((
@@ -335,9 +334,7 @@ pub async fn calculate_synthetic_size_worker(
 
                     if let Ok(tenant) = mgr::get_tenant(tenant_id, true).await
                     {
-                        if let Err(e) = tenant.calculate_synthetic_size(
-                            LogicalSizeCalculationCause::ConsumptionMetricsSyntheticSize,
-                            ctx).await {
+                        if let Err(e) = tenant.calculate_synthetic_size(ctx).await {
                             error!("failed to calculate synthetic size for tenant {}: {}", tenant_id, e);
                         }
                     }

pageserver/src/http/openapi_spec.yml

@@ -346,23 +346,23 @@ paths:
         starts writing to the tenant's S3 state unless it receives one of the
         distinguished errors below that state otherwise.
 
-        If a client receives a not-distinguished response, e.g., a network timeout,
-        it MUST retry the /attach request and poll again for the tenant's
-        attachment status.
-
-        After the client has received a 202, it MUST poll the tenant's
-        attachment status (field `attachment_status`) to reach state `attached`.
-        If the `attachment_status` is missing, the client MUST retry the `/attach`
-        request (goto previous paragraph). This is a robustness measure in case the tenant
-        status endpoint is buggy, but the attach operation is ongoing.
+        The method to identify whether a request has arrived at the pageserver, and
+        whether it has succeeded, is to poll for the tenant status to reach "Active"
+        or "Broken" state. These values are currently not explicitly documented in
+        the API spec.
+        Polling for `has_in_progress_downloads == false` is INCORRECT because that
+        value can turn `false` during shutdown while the Attach operation is still
+        unfinished.
 
         There is no way to cancel an in-flight request.
 
-        In any case, the client
-        * MUST NOT ASSUME that the /attach request has been lost in the network,
-        * MUST NOT ASSUME that the request has been lost, based on the observation
-          that a subsequent tenant status request returns 404. The request may
-          still be in flight. It must be retried.
+        If a client receives a not-distinguished response, e.g., a network timeout,
+        it MUST retry the /attach request and poll again for tenant status.
+
+        In any case, it must
+        * NOT ASSUME that the /attach request has been lost in the network,
+        * NOT ASSUME that the request has been lost based on a subsequent
+          tenant status request returning 404. (The request may still be in flight!)
       responses:
         "202":
           description: Tenant attaching scheduled
@@ -888,27 +888,13 @@ components:
       type: object
       required:
         - id
-        - attachment_status
       properties:
         id:
           type: string
         current_physical_size:
           type: integer
-        attachment_status:
-          description: |
-            Status of this tenant's attachment to this pageserver.
-
-            - `maybe` means almost nothing, don't read anything into it
-              except for the fact that the pageserver _might_ be already
-              writing to the tenant's S3 state, so, DO NOT ATTACH the
-              tenant to any other pageserver, or we risk split-brain.
-            - `attached` means that the attach operation has completed,
-              maybe successfully, maybe not. Perform a health check at
-              the Postgres level to determine healthiness of the tenant.
-
-            See the tenant `/attach` endpoint for more information.
-          type: string
-          enum: [ "maybe", "attached" ]
+        has_in_progress_downloads:
+          type: boolean
     TenantCreateInfo:
       type: object
       properties:

pageserver/src/http/routes.rs

@@ -25,7 +25,7 @@ use crate::tenant::config::TenantConfOpt;
 use crate::tenant::mgr::{TenantMapInsertError, TenantStateError};
 use crate::tenant::size::ModelInputs;
 use crate::tenant::storage_layer::LayerAccessStatsReset;
-use crate::tenant::{LogicalSizeCalculationCause, PageReconstructError, Timeline};
+use crate::tenant::{PageReconstructError, Timeline};
 use crate::{config::PageServerConf, tenant::mgr};
 use utils::{
     auth::JwtAuth,
@@ -105,9 +105,6 @@ impl From<PageReconstructError> for ApiError {
             PageReconstructError::Cancelled => {
                 ApiError::InternalServerError(anyhow::anyhow!("request was cancelled"))
             }
-            PageReconstructError::AncestorStopping(_) => {
-                ApiError::InternalServerError(anyhow::Error::new(pre))
-            }
             PageReconstructError::WalRedo(pre) => {
                 ApiError::InternalServerError(anyhow::Error::new(pre))
             }
@@ -172,8 +169,6 @@ async fn build_timeline_info(
     include_non_incremental_logical_size: bool,
     ctx: &RequestContext,
 ) -> anyhow::Result<TimelineInfo> {
-    crate::tenant::debug_assert_current_span_has_tenant_and_timeline_id();
-
     let mut info = build_timeline_info_common(timeline, ctx)?;
     if include_non_incremental_logical_size {
         // XXX we should be using spawn_ondemand_logical_size_calculation here.
@@ -196,7 +191,6 @@ fn build_timeline_info_common(
     timeline: &Arc<Timeline>,
     ctx: &RequestContext,
 ) -> anyhow::Result<TimelineInfo> {
-    crate::tenant::debug_assert_current_span_has_tenant_and_timeline_id();
     let last_record_lsn = timeline.get_last_record_lsn();
     let (wal_source_connstr, last_received_msg_lsn, last_received_msg_ts) = {
         let guard = timeline.last_received_wal.lock().unwrap();
@@ -269,28 +263,25 @@ async fn timeline_create_handler(mut request: Request<Body>) -> Result<Response<
 
     let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Error);
 
-    async {
-        let tenant = mgr::get_tenant(tenant_id, true).await?;
-        match tenant.create_timeline(
-            new_timeline_id,
-            request_data.ancestor_timeline_id.map(TimelineId::from),
-            request_data.ancestor_start_lsn,
-            request_data.pg_version.unwrap_or(crate::DEFAULT_PG_VERSION),
-            &ctx,
-        )
-        .await {
-            Ok(Some(new_timeline)) => {
-                // Created. Construct a TimelineInfo for it.
-                let timeline_info = build_timeline_info_common(&new_timeline, &ctx)
-                    .map_err(ApiError::InternalServerError)?;
-                json_response(StatusCode::CREATED, timeline_info)
-            }
-            Ok(None) => json_response(StatusCode::CONFLICT, ()), // timeline already exists
-            Err(err) => Err(ApiError::InternalServerError(err)),
-        }
-    }
+    let tenant = mgr::get_tenant(tenant_id, true).await?;
+    match tenant.create_timeline(
+        new_timeline_id,
+        request_data.ancestor_timeline_id.map(TimelineId::from),
+        request_data.ancestor_start_lsn,
+        request_data.pg_version.unwrap_or(crate::DEFAULT_PG_VERSION),
+        &ctx,
+    )
     .instrument(info_span!("timeline_create", tenant = %tenant_id, new_timeline = ?request_data.new_timeline_id, timeline_id = %new_timeline_id, lsn=?request_data.ancestor_start_lsn, pg_version=?request_data.pg_version))
-    .await
+    .await {
+        Ok(Some(new_timeline)) => {
+            // Created. Construct a TimelineInfo for it.
+            let timeline_info = build_timeline_info_common(&new_timeline, &ctx)
+                .map_err(ApiError::InternalServerError)?;
+            json_response(StatusCode::CREATED, timeline_info)
+        }
+        Ok(None) => json_response(StatusCode::CONFLICT, ()), // timeline already exists
+        Err(err) => Err(ApiError::InternalServerError(err)),
+    }
 }
 
 async fn timeline_list_handler(request: Request<Body>) -> Result<Response<Body>, ApiError> {
@@ -312,7 +303,6 @@ async fn timeline_list_handler(request: Request<Body>) -> Result<Response<Body>,
                 include_non_incremental_logical_size.unwrap_or(false),
                 &ctx,
             )
-            .instrument(info_span!("build_timeline_info", timeline_id = %timeline.timeline_id))
             .await
             .context("Failed to convert tenant timeline {timeline_id} into the local one: {e:?}")
             .map_err(ApiError::InternalServerError)?;
@@ -477,7 +467,7 @@ async fn tenant_list_handler(request: Request<Body>) -> Result<Response<Body>, A
             id: *id,
             state: state.clone(),
             current_physical_size: None,
-            attachment_status: state.attachment_status(),
+            has_in_progress_downloads: Some(state.has_in_progress_downloads()),
         })
         .collect::<Vec<TenantInfo>>();
 
@@ -502,7 +492,7 @@ async fn tenant_status(request: Request<Body>) -> Result<Response<Body>, ApiErro
             id: tenant_id,
             state: state.clone(),
             current_physical_size: Some(current_physical_size),
-            attachment_status: state.attachment_status(),
+            has_in_progress_downloads: Some(state.has_in_progress_downloads()),
         })
     }
     .instrument(info_span!("tenant_status_handler", tenant = %tenant_id))
@@ -537,11 +527,7 @@ async fn tenant_size_handler(request: Request<Body>) -> Result<Response<Body>, A
 
     // this can be long operation
     let inputs = tenant
-        .gather_size_inputs(
-            retention_period,
-            LogicalSizeCalculationCause::TenantSizeHandler,
-            &ctx,
-        )
+        .gather_size_inputs(retention_period, &ctx)
         .await
         .map_err(ApiError::InternalServerError)?;
 

pageserver/src/metrics.rs

@@ -30,7 +30,6 @@ const STORAGE_TIME_OPERATIONS: &[&str] = &[
     "create images",
     "init logical size",
     "logical size",
-    "imitate logical size",
     "load layer map",
     "gc",
 ];
@@ -187,16 +186,6 @@ static PERSISTENT_BYTES_WRITTEN: Lazy<IntCounterVec> = Lazy::new(|| {
     .expect("failed to define a metric")
 });
 
-pub(crate) static EVICTION_ITERATION_DURATION: Lazy<HistogramVec> = Lazy::new(|| {
-    register_histogram_vec!(
-        "pageserver_eviction_iteration_duration_seconds_global",
-        "Time spent on a single eviction iteration",
-        &["period_secs", "threshold_secs"],
-        STORAGE_OP_BUCKETS.into(),
-    )
-    .expect("failed to define a metric")
-});
-
 static EVICTIONS: Lazy<IntCounterVec> = Lazy::new(|| {
     register_int_counter_vec!(
         "pageserver_evictions",
@@ -489,15 +478,6 @@ pub static TENANT_TASK_EVENTS: Lazy<IntCounterVec> = Lazy::new(|| {
     .expect("Failed to register tenant_task_events metric")
 });
 
-pub static BACKGROUND_LOOP_PERIOD_OVERRUN_COUNT: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        "pageserver_background_loop_period_overrun_count",
-        "Incremented whenever warn_when_period_overrun() logs a warning.",
-        &["task", "period"],
-    )
-    .expect("failed to define a metric")
-});
-
 // walreceiver metrics
 
 pub static WALRECEIVER_STARTED_CONNECTIONS: Lazy<IntCounter> = Lazy::new(|| {
@@ -708,7 +688,6 @@ pub struct TimelineMetrics {
     pub compact_time_histo: StorageTimeMetrics,
     pub create_images_time_histo: StorageTimeMetrics,
     pub logical_size_histo: StorageTimeMetrics,
-    pub imitate_logical_size_histo: StorageTimeMetrics,
     pub load_layer_map_histo: StorageTimeMetrics,
     pub garbage_collect_histo: StorageTimeMetrics,
     pub last_record_gauge: IntGauge,
@@ -741,8 +720,6 @@ impl TimelineMetrics {
         let create_images_time_histo =
             StorageTimeMetrics::new("create images", &tenant_id, &timeline_id);
         let logical_size_histo = StorageTimeMetrics::new("logical size", &tenant_id, &timeline_id);
-        let imitate_logical_size_histo =
-            StorageTimeMetrics::new("imitate logical size", &tenant_id, &timeline_id);
         let load_layer_map_histo =
             StorageTimeMetrics::new("load layer map", &tenant_id, &timeline_id);
         let garbage_collect_histo = StorageTimeMetrics::new("gc", &tenant_id, &timeline_id);
@@ -779,7 +756,6 @@ impl TimelineMetrics {
             compact_time_histo,
             create_images_time_histo,
             logical_size_histo,
-            imitate_logical_size_histo,
             garbage_collect_histo,
             load_layer_map_histo,
             last_record_gauge,
@@ -1240,7 +1216,4 @@ pub fn preinitialize_metrics() {
     // Initialize it eagerly, so that our alert rule can distinguish absence of the metric from metric value 0.
     assert_eq!(UNEXPECTED_ONDEMAND_DOWNLOADS.get(), 0);
     UNEXPECTED_ONDEMAND_DOWNLOADS.reset();
-
-    // Same as above for this metric, but, it's a Vec-type metric for which we don't know all the labels.
-    BACKGROUND_LOOP_PERIOD_OVERRUN_COUNT.reset();
 }

pageserver/src/page_service.rs

@@ -256,10 +256,7 @@ async fn page_service_conn_main(
     //
     // no write timeout is used, because the kernel is assumed to error writes after some time.
     let mut socket = tokio_io_timeout::TimeoutReader::new(socket);
-
-    // timeout should be lower, but trying out multiple days for
-    // <https://github.com/neondatabase/neon/issues/4205>
-    socket.set_timeout(Some(std::time::Duration::from_secs(60 * 60 * 24 * 3)));
+    socket.set_timeout(Some(std::time::Duration::from_secs(60 * 10)));
     let socket = std::pin::pin!(socket);
 
     // XXX: pgbackend.run() should take the connection_ctx,

pageserver/src/pgdatadir_mapping.rs

@@ -500,8 +500,6 @@ impl Timeline {
         cancel: CancellationToken,
         ctx: &RequestContext,
     ) -> Result<u64, CalculateLogicalSizeError> {
-        crate::tenant::debug_assert_current_span_has_tenant_and_timeline_id();
-
         // Fetch list of database dirs and iterate them
         let buf = self.get(DBDIR_KEY, lsn, ctx).await.context("read dbdir")?;
         let dbdir = DbDirectory::des(&buf).context("deserialize db directory")?;

pageserver/src/task_mgr.rs

@@ -272,6 +272,9 @@ pub enum TaskKind {
 
     #[cfg(test)]
     UnitTest,
+
+    /// Task which is the only task to delete this particular timeline
+    DeleteTimeline,
 }
 
 #[derive(Default)]

pageserver/src/tenant.rs

@@ -97,10 +97,7 @@ mod timeline;
 
 pub mod size;
 
-pub(crate) use timeline::debug_assert_current_span_has_tenant_and_timeline_id;
-pub use timeline::{
-    LocalLayerInfoForDiskUsageEviction, LogicalSizeCalculationCause, PageReconstructError, Timeline,
-};
+pub use timeline::{LocalLayerInfoForDiskUsageEviction, PageReconstructError, Timeline};
 
 // re-export this function so that page_cache.rs can use it.
 pub use crate::tenant::ephemeral_file::writeback as writeback_ephemeral_file;
@@ -459,6 +456,50 @@ pub enum DeleteTimelineError {
     Other(#[from] anyhow::Error),
 }
 
+#[derive(Debug, thiserror::Error, Clone)]
+enum InnerDeleteTimelineError {
+    #[error("upload queue is uninitialized, likely the timeline was in Broken state prior to this call because it failed to fetch IndexPart during load or attach, check the logs")]
+    QueueUninitialized,
+
+    #[error("failpoint: {0}")]
+    Failpoint(&'static str),
+
+    #[error("failed to remove local timeline directory")]
+    FailedToRemoveLocalTimelineDirectory,
+
+    #[error("index_part.json upload failed")]
+    UploadFailed,
+
+    #[error("the deleted timeline grew branches while deleting it; tenant should now be broken")]
+    DeletedGrewChildren,
+}
+
+impl utils::shared_retryable::Retryable for InnerDeleteTimelineError {
+    fn is_permanent(&self) -> bool {
+        use InnerDeleteTimelineError::*;
+
+        match self {
+            QueueUninitialized => false,
+            Failpoint(_) => false,
+            FailedToRemoveLocalTimelineDirectory => false,
+            UploadFailed => false,
+            DeletedGrewChildren => true,
+        }
+    }
+}
+
+impl From<InnerDeleteTimelineError> for DeleteTimelineError {
+    fn from(value: InnerDeleteTimelineError) -> Self {
+        DeleteTimelineError::Other(anyhow::Error::new(value))
+    }
+}
+
+impl From<utils::shared_retryable::RetriedTaskPanicked> for DeleteTimelineError {
+    fn from(_: utils::shared_retryable::RetriedTaskPanicked) -> Self {
+        DeleteTimelineError::Other(anyhow::anyhow!("deleting timeline failed, please retry"))
+    }
+}
+
 struct RemoteStartupData {
     index_part: IndexPart,
     remote_metadata: TimelineMetadata,
@@ -596,19 +637,16 @@ impl Tenant {
     /// finishes. You can use wait_until_active() to wait for the task to
     /// complete.
     ///
-    pub(crate) fn spawn_attach(
+    pub fn spawn_attach(
         conf: &'static PageServerConf,
         tenant_id: TenantId,
         remote_storage: GenericRemoteStorage,
         ctx: &RequestContext,
-    ) -> anyhow::Result<Arc<Tenant>> {
+    ) -> Arc<Tenant> {
         // XXX: Attach should provide the config, especially during tenant migration.
         //      See https://github.com/neondatabase/neon/issues/1555
         let tenant_conf = TenantConfOpt::default();
 
-        Self::attach_idempotent_create_marker_file(conf, tenant_id)
-            .context("create attach marker file")?;
-
         let wal_redo_manager = Arc::new(PostgresRedoManager::new(conf, tenant_id));
         let tenant = Arc::new(Tenant::new(
             TenantState::Attaching,
@@ -641,46 +679,7 @@ impl Tenant {
                 Ok(())
             },
         );
-        Ok(tenant)
-    }
-
-    fn attach_idempotent_create_marker_file(
-        conf: &'static PageServerConf,
-        tenant_id: TenantId,
-    ) -> anyhow::Result<()> {
-        // Create directory with marker file to indicate attaching state.
-        // The load_local_tenants() function in tenant::mgr relies on the marker file
-        // to determine whether a tenant has finished attaching.
-        let tenant_dir = conf.tenant_path(&tenant_id);
-        let marker_file = conf.tenant_attaching_mark_file_path(&tenant_id);
-        debug_assert_eq!(marker_file.parent().unwrap(), tenant_dir);
-        // TODO: should use tokio::fs here, but
-        // 1. caller is not async, for good reason (it holds tenants map lock)
-        // 2. we'd need to think about cancel safety. Turns out dropping a tokio::fs future
-        //    doesn't wait for the activity in the fs thread pool.
-        crashsafe::create_dir_all(&tenant_dir).context("create tenant directory")?;
-        match fs::OpenOptions::new()
-            .write(true)
-            .create_new(true)
-            .open(&marker_file)
-        {
-            Ok(_) => {}
-            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
-                // Either this is a retry of attach or there is a concurrent task also doing attach for this tenant.
-                // We cannot distinguish this here.
-                // The caller is responsible for ensuring there's no concurrent attach for a tenant.
-                {} // fsync again, we don't know if that already happened
-            }
-            err => {
-                err.context("create tenant attaching marker file")?;
-                unreachable!("we covered the Ok() case above");
-            }
-        }
-        crashsafe::fsync_file_and_parent(&marker_file)
-            .context("fsync tenant attaching marker file and parent")?;
-        debug_assert!(tenant_dir.is_dir());
-        debug_assert!(marker_file.is_file());
-        Ok(())
+        tenant
     }
 
     ///
@@ -688,15 +687,26 @@ impl Tenant {
     ///
     #[instrument(skip_all, fields(tenant_id=%self.tenant_id))]
     async fn attach(self: &Arc<Tenant>, ctx: RequestContext) -> anyhow::Result<()> {
+        // Create directory with marker file to indicate attaching state.
+        // The load_local_tenants() function in tenant::mgr relies on the marker file
+        // to determine whether a tenant has finished attaching.
+        let tenant_dir = self.conf.tenant_path(&self.tenant_id);
         let marker_file = self.conf.tenant_attaching_mark_file_path(&self.tenant_id);
-        if !tokio::fs::try_exists(&marker_file)
-            .await
-            .context("check for existence of marker file")?
-        {
-            anyhow::bail!(
-                "implementation error: marker file should exist at beginning of this function"
-            );
+        debug_assert_eq!(marker_file.parent().unwrap(), tenant_dir);
+        if tenant_dir.exists() {
+            if !marker_file.is_file() {
+                anyhow::bail!(
+                    "calling Tenant::attach with a tenant directory that doesn't have the attaching marker file:\ntenant_dir: {}\nmarker_file: {}",
+                    tenant_dir.display(), marker_file.display());
+            }
+        } else {
+            crashsafe::create_dir_all(&tenant_dir).context("create tenant directory")?;
+            fs::File::create(&marker_file).context("create tenant attaching marker file")?;
+            crashsafe::fsync_file_and_parent(&marker_file)
+                .context("fsync tenant attaching marker file and parent")?;
         }
+        debug_assert!(tenant_dir.is_dir());
+        debug_assert!(marker_file.is_file());
 
         // Get list of remote timelines
         // download index files for every tenant timeline
@@ -873,15 +883,11 @@ impl Tenant {
     }
 
     /// Create a placeholder Tenant object for a broken tenant
-    pub fn create_broken_tenant(
-        conf: &'static PageServerConf,
-        tenant_id: TenantId,
-        reason: String,
-    ) -> Arc<Tenant> {
+    pub fn create_broken_tenant(conf: &'static PageServerConf, tenant_id: TenantId) -> Arc<Tenant> {
         let wal_redo_manager = Arc::new(PostgresRedoManager::new(conf, tenant_id));
         Arc::new(Tenant::new(
             TenantState::Broken {
-                reason,
+                reason: "create_broken_tenant".into(),
                 backtrace: String::new(),
             },
             conf,
@@ -914,7 +920,7 @@ impl Tenant {
             Ok(conf) => conf,
             Err(e) => {
                 error!("load tenant config failed: {:?}", e);
-                return Tenant::create_broken_tenant(conf, tenant_id, format!("{e:#}"));
+                return Tenant::create_broken_tenant(conf, tenant_id);
             }
         };
 
@@ -1264,24 +1270,8 @@ impl Tenant {
             "Cannot create timelines on inactive tenant"
         );
 
-        if let Ok(existing) = self.get_timeline(new_timeline_id, false) {
+        if self.get_timeline(new_timeline_id, false).is_ok() {
             debug!("timeline {new_timeline_id} already exists");
-
-            if let Some(remote_client) = existing.remote_client.as_ref() {
-                // Wait for uploads to complete, so that when we return Ok, the timeline
-                // is known to be durable on remote storage. Just like we do at the end of
-                // this function, after we have created the timeline ourselves.
-                //
-                // We only really care that the initial version of `index_part.json` has
-                // been uploaded. That's enough to remember that the timeline
-                // exists. However, there is no function to wait specifically for that so
-                // we just wait for all in-progress uploads to finish.
-                remote_client
-                    .wait_completion()
-                    .await
-                    .context("wait for timeline uploads to complete")?;
-            }
-
             return Ok(None);
         }
 
@@ -1323,17 +1313,6 @@ impl Tenant {
             }
         };
 
-        if let Some(remote_client) = loaded_timeline.remote_client.as_ref() {
-            // Wait for the upload of the 'index_part.json` file to finish, so that when we return
-            // Ok, the timeline is durable in remote storage.
-            let kind = ancestor_timeline_id
-                .map(|_| "branched")
-                .unwrap_or("bootstrapped");
-            remote_client.wait_completion().await.with_context(|| {
-                format!("wait for {} timeline initial uploads to complete", kind)
-            })?;
-        }
-
         Ok(Some(loaded_timeline))
     }
 
@@ -1426,7 +1405,7 @@ impl Tenant {
 
     /// Removes timeline-related in-memory data
     pub async fn delete_timeline(
-        &self,
+        self: &Arc<Tenant>,
         timeline_id: TimelineId,
         _ctx: &RequestContext,
     ) -> Result<(), DeleteTimelineError> {
@@ -1459,162 +1438,196 @@ impl Tenant {
             timeline
         };
 
-        // Now that the Timeline is in Stopping state, request all the related tasks to
-        // shut down.
+        let span = tracing::Span::current();
+
+        // if we have concurrent requests, we will only execute one version of following future;
+        // initially it did not have any means to be cancelled.
         //
-        // NB: If you call delete_timeline multiple times concurrently, they will
-        // all go through the motions here. Make sure the code here is idempotent,
-        // and don't error out if some of the shutdown tasks have already been
-        // completed!
+        // NOTE: Unlike "the usual" futures, this one will log any errors instead of just propagating
+        // them to the caller. This is because this one future produces a value, which will need to
+        // be cloneable to everyone interested, and normal `std::error::Error` are not clonable.
+        // Also, it wouldn't make sense to log the same failure multiple times, it would look like
+        // multiple failures to anyone reading the logs.
+        let factory = || {
+            let tenant = self.clone();
+            let tenant_id = self.tenant_id;
+            let timeline = timeline.clone();
+            let timeline_id = timeline.timeline_id;
 
-        // Stop the walreceiver first.
-        debug!("waiting for wal receiver to shutdown");
-        timeline.walreceiver.stop().await;
-        debug!("wal receiver shutdown confirmed");
+            async move {
+                // Now that the Timeline is in Stopping state, request all the related tasks to
+                // shut down.
+                //
+                // Stop the walreceiver first.
+                debug!("waiting for wal receiver to shutdown");
+                timeline.walreceiver.stop().await;
+                debug!("wal receiver shutdown confirmed");
 
-        // Prevent new uploads from starting.
-        if let Some(remote_client) = timeline.remote_client.as_ref() {
-            let res = remote_client.stop();
-            match res {
-                Ok(()) => {}
-                Err(e) => match e {
-                    remote_timeline_client::StopError::QueueUninitialized => {
-                        // This case shouldn't happen currently because the
-                        // load and attach code bails out if _any_ of the timeline fails to fetch its IndexPart.
-                        // That is, before we declare the Tenant as Active.
-                        // But we only allow calls to delete_timeline on Active tenants.
-                        return Err(DeleteTimelineError::Other(anyhow::anyhow!("upload queue is uninitialized, likely the timeline was in Broken state prior to this call because it failed to fetch IndexPart during load or attach, check the logs")));
+                // Prevent new uploads from starting.
+                if let Some(remote_client) = timeline.remote_client.as_ref() {
+                    let res = remote_client.stop();
+                    match res {
+                        Ok(()) => {}
+                        Err(e) => match e {
+                            remote_timeline_client::StopError::QueueUninitialized => {
+                                // This case shouldn't happen currently because the
+                                // load and attach code bails out if _any_ of the timeline fails to fetch its IndexPart.
+                                // That is, before we declare the Tenant as Active.
+                                // But we only allow calls to delete_timeline on Active tenants.
+                                warn!("failed to stop RemoteTimelineClient due to uninitialized queue");
+                                return Err(InnerDeleteTimelineError::QueueUninitialized);
+                            }
+                        },
                     }
+                }
+
+                // Stop & wait for the remaining timeline tasks, including upload tasks.
+                info!("waiting for timeline tasks to shutdown");
+                task_mgr::shutdown_tasks(None, Some(tenant_id), Some(timeline_id)).await;
+
+                // Mark timeline as deleted in S3 so we won't pick it up next time
+                // during attach or pageserver restart.
+                // See comment in persist_index_part_with_deleted_flag.
+                if let Some(remote_client) = timeline.remote_client.as_ref() {
+                    match remote_client.persist_index_part_with_deleted_flag().await {
+                        // If we (now, or already) marked it successfully as deleted, we can proceed
+                        Ok(()) | Err(PersistIndexPartWithDeletedFlagError::AlreadyDeleted(_)) => (),
+                        // Bail out otherwise
+                        Err(e @ PersistIndexPartWithDeletedFlagError::AlreadyInProgress(_))
+                        | Err(e @ PersistIndexPartWithDeletedFlagError::Other(_)) => {
+                            warn!("upload failed: {e}");
+                            return Err(InnerDeleteTimelineError::UploadFailed);
+                        }
+                    }
+                }
+
+                {
+                    // Grab the layer_removal_cs lock, and actually perform the deletion.
+                    //
+                    // This lock prevents multiple concurrent delete_timeline calls from
+                    // stepping on each other's toes, while deleting the files. It also
+                    // prevents GC or compaction from running at the same time.
+                    //
+                    // Note that there are still other race conditions between
+                    // GC, compaction and timeline deletion. GC task doesn't
+                    // register itself properly with the timeline it's
+                    // operating on. See
+                    // https://github.com/neondatabase/neon/issues/2671
+                    //
+                    // No timeout here, GC & Compaction should be responsive to the
+                    // `TimelineState::Stopping` change.
+                    info!("waiting for layer_removal_cs.lock()");
+                    let _layer_removal_guard = timeline.layer_removal_cs.lock().await;
+                    info!("got layer_removal_cs.lock(), deleting layer files");
+
+                    // NB: storage_sync upload tasks that reference these layers have been cancelled
+                    //     by us earlier.
+
+                    let local_timeline_directory =
+                        tenant.conf.timeline_path(&timeline_id, &tenant_id);
+
+                    fail::fail_point!("timeline-delete-before-rm", |_| {
+                        Err(InnerDeleteTimelineError::Failpoint(
+                            "failpoint: timeline-delete-before-rm",
+                        ))
+                    });
+
+                    // NB: This need not be atomic because the deleted flag in the IndexPart
+                    // will be observed during tenant/timeline load. The deletion will be resumed there.
+                    //
+                    // For configurations without remote storage, we tolerate that we're not crash-safe here.
+                    // The timeline may come up Active but with missing layer files, in such setups.
+                    // See https://github.com/neondatabase/neon/pull/3919#issuecomment-1531726720
+                    match std::fs::remove_dir_all(&local_timeline_directory) {
+                        Ok(()) => {}
+                        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
+                            // This can happen if we're called a second time, e.g.,
+                            // because of a previous failure/cancellation at/after
+                            // failpoint timeline-delete-after-rm.
+                            //
+                            // It can also happen if we race with tenant detach, because,
+                            // it doesn't grab the layer_removal_cs lock.
+                            //
+                            // For now, log and continue.
+                            // warn! level is technically not appropriate for the
+                            // first case because we should expect retries to happen.
+                            // But the error is so rare, it seems better to get attention if it happens.
+                            let tenant_state = tenant.current_state();
+                            warn!(
+                                timeline_dir=?local_timeline_directory,
+                                ?tenant_state,
+                                "timeline directory not found, proceeding anyway"
+                            );
+                        }
+                        Err(e) => {
+                            warn!(
+                                "failed to remove local timeline directory {}: {e:#}",
+                                local_timeline_directory.display()
+                            );
+                            return Err(
+                                InnerDeleteTimelineError::FailedToRemoveLocalTimelineDirectory,
+                            );
+                        }
+                    }
+
+                    info!("finished deleting layer files, releasing layer_removal_cs.lock()");
+                }
+
+                fail::fail_point!("timeline-delete-after-rm", |_| {
+                    Err(InnerDeleteTimelineError::Failpoint(
+                        "timeline-delete-after-rm",
+                    ))
+                });
+
+                // Remove the timeline from the map or poison it if we've grown children.
+                let removed_timeline =
+                    std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+                        let mut timelines = tenant.timelines.lock().unwrap();
+                        let children_exist = timelines.iter().any(|(_, entry)| {
+                            entry.get_ancestor_timeline_id() == Some(timeline_id)
+                        });
+                        // XXX this can happen because `branch_timeline` doesn't check `TimelineState::Stopping`.
+                        // We already deleted the layer files, so it's probably best to panic.
+                        // (Ideally, above remove_dir_all is atomic so we don't see this timeline after a restart)
+                        if children_exist {
+                            panic!("Timeline grew children while we removed layer files");
+                        }
+                        timelines.remove(&timeline_id)
+                    }));
+
+                match removed_timeline {
+                    Ok(Some(_)) => {}
+                    Ok(None) => {
+                        // with SharedRetryable this should no longer happen
+                        warn!("no other task should had dropped the Timeline");
+                    }
+                    Err(_panic) => return Err(InnerDeleteTimelineError::DeletedGrewChildren),
+                }
+
+                Ok(())
+            }
+            // execute in the *winner's* span so we will capture the request id etc.
+            .instrument(span)
+        };
+
+        let (recv, maybe_fut) = timeline.delete_self.try_restart(factory).await;
+
+        if let Some(fut) = maybe_fut {
+            crate::task_mgr::spawn(
+                &tokio::runtime::Handle::current(),
+                TaskKind::DeleteTimeline,
+                Some(self.tenant_id()),
+                None,
+                &format!("delete_timeline {}", timeline.timeline_id),
+                false,
+                async move {
+                    fut.await;
+                    Ok(())
                 },
-            }
+            );
         }
 
-        // Stop & wait for the remaining timeline tasks, including upload tasks.
-        // NB: This and other delete_timeline calls do not run as a task_mgr task,
-        //     so, they are not affected by this shutdown_tasks() call.
-        info!("waiting for timeline tasks to shutdown");
-        task_mgr::shutdown_tasks(None, Some(self.tenant_id), Some(timeline_id)).await;
-
-        // Mark timeline as deleted in S3 so we won't pick it up next time
-        // during attach or pageserver restart.
-        // See comment in persist_index_part_with_deleted_flag.
-        if let Some(remote_client) = timeline.remote_client.as_ref() {
-            match remote_client.persist_index_part_with_deleted_flag().await {
-                // If we (now, or already) marked it successfully as deleted, we can proceed
-                Ok(()) | Err(PersistIndexPartWithDeletedFlagError::AlreadyDeleted(_)) => (),
-                // Bail out otherwise
-                Err(e @ PersistIndexPartWithDeletedFlagError::AlreadyInProgress(_))
-                | Err(e @ PersistIndexPartWithDeletedFlagError::Other(_)) => {
-                    return Err(DeleteTimelineError::Other(anyhow::anyhow!(e)));
-                }
-            }
-        }
-
-        {
-            // Grab the layer_removal_cs lock, and actually perform the deletion.
-            //
-            // This lock prevents multiple concurrent delete_timeline calls from
-            // stepping on each other's toes, while deleting the files. It also
-            // prevents GC or compaction from running at the same time.
-            //
-            // Note that there are still other race conditions between
-            // GC, compaction and timeline deletion. GC task doesn't
-            // register itself properly with the timeline it's
-            // operating on. See
-            // https://github.com/neondatabase/neon/issues/2671
-            //
-            // No timeout here, GC & Compaction should be responsive to the
-            // `TimelineState::Stopping` change.
-            info!("waiting for layer_removal_cs.lock()");
-            let layer_removal_guard = timeline.layer_removal_cs.lock().await;
-            info!("got layer_removal_cs.lock(), deleting layer files");
-
-            // NB: storage_sync upload tasks that reference these layers have been cancelled
-            //     by the caller.
-
-            let local_timeline_directory = self.conf.timeline_path(&timeline_id, &self.tenant_id);
-
-            fail::fail_point!("timeline-delete-before-rm", |_| {
-                Err(anyhow::anyhow!("failpoint: timeline-delete-before-rm"))?
-            });
-
-            // NB: This need not be atomic because the deleted flag in the IndexPart
-            // will be observed during tenant/timeline load. The deletion will be resumed there.
-            //
-            // For configurations without remote storage, we tolerate that we're not crash-safe here.
-            // The timeline may come up Active but with missing layer files, in such setups.
-            // See https://github.com/neondatabase/neon/pull/3919#issuecomment-1531726720
-            match std::fs::remove_dir_all(&local_timeline_directory) {
-                Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
-                    // This can happen if we're called a second time, e.g.,
-                    // because of a previous failure/cancellation at/after
-                    // failpoint timeline-delete-after-rm.
-                    //
-                    // It can also happen if we race with tenant detach, because,
-                    // it doesn't grab the layer_removal_cs lock.
-                    //
-                    // For now, log and continue.
-                    // warn! level is technically not appropriate for the
-                    // first case because we should expect retries to happen.
-                    // But the error is so rare, it seems better to get attention if it happens.
-                    let tenant_state = self.current_state();
-                    warn!(
-                        timeline_dir=?local_timeline_directory,
-                        ?tenant_state,
-                        "timeline directory not found, proceeding anyway"
-                    );
-                    // continue with the rest of the deletion
-                }
-                res => res.with_context(|| {
-                    format!(
-                        "Failed to remove local timeline directory '{}'",
-                        local_timeline_directory.display()
-                    )
-                })?,
-            }
-
-            info!("finished deleting layer files, releasing layer_removal_cs.lock()");
-            drop(layer_removal_guard);
-        }
-
-        fail::fail_point!("timeline-delete-after-rm", |_| {
-            Err(anyhow::anyhow!("failpoint: timeline-delete-after-rm"))?
-        });
-
-        // Remove the timeline from the map.
-        let mut timelines = self.timelines.lock().unwrap();
-        let children_exist = timelines
-            .iter()
-            .any(|(_, entry)| entry.get_ancestor_timeline_id() == Some(timeline_id));
-        // XXX this can happen because `branch_timeline` doesn't check `TimelineState::Stopping`.
-        // We already deleted the layer files, so it's probably best to panic.
-        // (Ideally, above remove_dir_all is atomic so we don't see this timeline after a restart)
-        if children_exist {
-            panic!("Timeline grew children while we removed layer files");
-        }
-        let removed_timeline = timelines.remove(&timeline_id);
-        if removed_timeline.is_none() {
-            // This can legitimately happen if there's a concurrent call to this function.
-            //   T1                                             T2
-            //   lock
-            //   unlock
-            //                                                  lock
-            //                                                  unlock
-            //                                                  remove files
-            //                                                  lock
-            //                                                  remove from map
-            //                                                  unlock
-            //                                                  return
-            //   remove files
-            //   lock
-            //   remove from map observes empty map
-            //   unlock
-            //   return
-            debug!("concurrent call to this function won the race");
-        }
-        drop(timelines);
-
-        Ok(())
+        recv.await
     }
 
     pub fn current_state(&self) -> TenantState {
@@ -2403,18 +2416,17 @@ impl Tenant {
             src_timeline.initdb_lsn,
             src_timeline.pg_version,
         );
-
-        let new_timeline = {
-            let mut timelines = self.timelines.lock().unwrap();
-            self.prepare_timeline(
+        let mut timelines = self.timelines.lock().unwrap();
+        let new_timeline = self
+            .prepare_timeline(
                 dst_id,
                 &metadata,
                 timeline_uninit_mark,
                 false,
                 Some(Arc::clone(src_timeline)),
             )?
-            .initialize_with_lock(ctx, &mut timelines, true, true)?
-        };
+            .initialize_with_lock(ctx, &mut timelines, true, true)?;
+        drop(timelines);
 
         // Root timeline gets its layers during creation and uploads them along with the metadata.
         // A branch timeline though, when created, can get no writes for some time, hence won't get any layers created.
@@ -2672,7 +2684,6 @@ impl Tenant {
         // `max_retention_period` overrides the cutoff that is used to calculate the size
         // (only if it is shorter than the real cutoff).
         max_retention_period: Option<u64>,
-        cause: LogicalSizeCalculationCause,
         ctx: &RequestContext,
     ) -> anyhow::Result<size::ModelInputs> {
         let logical_sizes_at_once = self
@@ -2694,7 +2705,6 @@ impl Tenant {
             logical_sizes_at_once,
             max_retention_period,
             &mut shared_cache,
-            cause,
             ctx,
         )
         .await
@@ -2704,12 +2714,8 @@ impl Tenant {
     /// This is periodically called by background worker.
     /// result is cached in tenant struct
     #[instrument(skip_all, fields(tenant_id=%self.tenant_id))]
-    pub async fn calculate_synthetic_size(
-        &self,
-        cause: LogicalSizeCalculationCause,
-        ctx: &RequestContext,
-    ) -> anyhow::Result<u64> {
-        let inputs = self.gather_size_inputs(None, cause, ctx).await?;
+    pub async fn calculate_synthetic_size(&self, ctx: &RequestContext) -> anyhow::Result<u64> {
+        let inputs = self.gather_size_inputs(None, ctx).await?;
 
         let size = inputs.calculate()?;
 
@@ -3469,26 +3475,14 @@ mod tests {
             .gc_iteration(Some(TIMELINE_ID), 0x10, Duration::ZERO, &ctx)
             .await?;
 
-        // The branchpoints should contain all timelines, even ones marked
-        // as Broken.
-        {
-            let branchpoints = &tline.gc_info.read().unwrap().retain_lsns;
-            assert_eq!(branchpoints.len(), 1);
-            assert_eq!(branchpoints[0], Lsn(0x40));
-        }
-
-        // You can read the key from the child branch even though the parent is
-        // Broken, as long as you don't need to access data from the parent.
         assert_eq!(
-            newtline.get(*TEST_KEY, Lsn(0x70), &ctx).await?,
-            TEST_IMG(&format!("foo at {}", Lsn(0x70)))
+            newtline.get(*TEST_KEY, Lsn(0x50), &ctx).await?,
+            TEST_IMG(&format!("foo at {}", Lsn(0x40)))
         );
 
-        // This needs to traverse to the parent, and fails.
-        let err = newtline.get(*TEST_KEY, Lsn(0x50), &ctx).await.unwrap_err();
-        assert!(err
-            .to_string()
-            .contains("will not become active. Current state: Broken"));
+        let branchpoints = &tline.gc_info.read().unwrap().retain_lsns;
+        assert_eq!(branchpoints.len(), 1);
+        assert_eq!(branchpoints[0], Lsn(0x40));
 
         Ok(())
     }

pageserver/src/tenant/config.rs

@@ -292,93 +292,77 @@ fn bad_duration<'a>(field_name: &'static str, value: &'a str) -> impl 'a + Fn()
     move || format!("Cannot parse `{field_name}` duration {value:?}")
 }
 
-impl TenantConfOpt {
-    #[allow(clippy::too_many_arguments)]
-    fn from_request(
-        checkpoint_distance: Option<u64>,
-        checkpoint_timeout: &Option<String>,
-        compaction_target_size: Option<u64>,
-        compaction_period: &Option<String>,
-        compaction_threshold: Option<usize>,
-        gc_horizon: Option<u64>,
-        gc_period: &Option<String>,
-        image_creation_threshold: Option<usize>,
-        pitr_interval: &Option<String>,
-        walreceiver_connect_timeout: &Option<String>,
-        lagging_wal_timeout: &Option<String>,
-        max_lsn_wal_lag: Option<NonZeroU64>,
-        trace_read_requests: Option<bool>,
-        eviction_policy: &Option<serde_json::Value>,
-        min_resident_size_override: Option<u64>,
-        evictions_low_residence_duration_metric_threshold: &Option<String>,
-    ) -> Result<Self, anyhow::Error> {
+impl TryFrom<&'_ TenantCreateRequest> for TenantConfOpt {
+    type Error = anyhow::Error;
+
+    fn try_from(request_data: &TenantCreateRequest) -> Result<Self, Self::Error> {
         let mut tenant_conf = TenantConfOpt::default();
 
-        if let Some(gc_period) = &gc_period {
+        if let Some(gc_period) = &request_data.gc_period {
             tenant_conf.gc_period = Some(
                 humantime::parse_duration(gc_period)
                     .with_context(bad_duration("gc_period", gc_period))?,
             );
         }
-        tenant_conf.gc_horizon = gc_horizon;
-        tenant_conf.image_creation_threshold = image_creation_threshold;
+        tenant_conf.gc_horizon = request_data.gc_horizon;
+        tenant_conf.image_creation_threshold = request_data.image_creation_threshold;
 
-        if let Some(pitr_interval) = &pitr_interval {
+        if let Some(pitr_interval) = &request_data.pitr_interval {
             tenant_conf.pitr_interval = Some(
                 humantime::parse_duration(pitr_interval)
                     .with_context(bad_duration("pitr_interval", pitr_interval))?,
             );
         }
 
-        if let Some(walreceiver_connect_timeout) = &walreceiver_connect_timeout {
+        if let Some(walreceiver_connect_timeout) = &request_data.walreceiver_connect_timeout {
             tenant_conf.walreceiver_connect_timeout = Some(
                 humantime::parse_duration(walreceiver_connect_timeout).with_context(
                     bad_duration("walreceiver_connect_timeout", walreceiver_connect_timeout),
                 )?,
             );
         }
-        if let Some(lagging_wal_timeout) = &lagging_wal_timeout {
+        if let Some(lagging_wal_timeout) = &request_data.lagging_wal_timeout {
             tenant_conf.lagging_wal_timeout = Some(
                 humantime::parse_duration(lagging_wal_timeout)
                     .with_context(bad_duration("lagging_wal_timeout", lagging_wal_timeout))?,
             );
         }
-        if let Some(max_lsn_wal_lag) = max_lsn_wal_lag {
+        if let Some(max_lsn_wal_lag) = request_data.max_lsn_wal_lag {
             tenant_conf.max_lsn_wal_lag = Some(max_lsn_wal_lag);
         }
-        if let Some(trace_read_requests) = trace_read_requests {
+        if let Some(trace_read_requests) = request_data.trace_read_requests {
             tenant_conf.trace_read_requests = Some(trace_read_requests);
         }
 
-        tenant_conf.checkpoint_distance = checkpoint_distance;
-        if let Some(checkpoint_timeout) = &checkpoint_timeout {
+        tenant_conf.checkpoint_distance = request_data.checkpoint_distance;
+        if let Some(checkpoint_timeout) = &request_data.checkpoint_timeout {
             tenant_conf.checkpoint_timeout = Some(
                 humantime::parse_duration(checkpoint_timeout)
                     .with_context(bad_duration("checkpoint_timeout", checkpoint_timeout))?,
             );
         }
 
-        tenant_conf.compaction_target_size = compaction_target_size;
-        tenant_conf.compaction_threshold = compaction_threshold;
+        tenant_conf.compaction_target_size = request_data.compaction_target_size;
+        tenant_conf.compaction_threshold = request_data.compaction_threshold;
 
-        if let Some(compaction_period) = &compaction_period {
+        if let Some(compaction_period) = &request_data.compaction_period {
             tenant_conf.compaction_period = Some(
                 humantime::parse_duration(compaction_period)
                     .with_context(bad_duration("compaction_period", compaction_period))?,
             );
         }
 
-        if let Some(eviction_policy) = &eviction_policy {
+        if let Some(eviction_policy) = &request_data.eviction_policy {
             tenant_conf.eviction_policy = Some(
                 serde::Deserialize::deserialize(eviction_policy)
                     .context("parse field `eviction_policy`")?,
             );
         }
 
-        tenant_conf.min_resident_size_override = min_resident_size_override;
+        tenant_conf.min_resident_size_override = request_data.min_resident_size_override;
 
         if let Some(evictions_low_residence_duration_metric_threshold) =
-            &evictions_low_residence_duration_metric_threshold
+            &request_data.evictions_low_residence_duration_metric_threshold
         {
             tenant_conf.evictions_low_residence_duration_metric_threshold = Some(
                 humantime::parse_duration(evictions_low_residence_duration_metric_threshold)
@@ -393,53 +377,81 @@ impl TenantConfOpt {
     }
 }
 
-impl TryFrom<&'_ TenantCreateRequest> for TenantConfOpt {
-    type Error = anyhow::Error;
-
-    fn try_from(request_data: &TenantCreateRequest) -> Result<Self, Self::Error> {
-        Self::from_request(
-            request_data.checkpoint_distance,
-            &request_data.checkpoint_timeout,
-            request_data.compaction_target_size,
-            &request_data.compaction_period,
-            request_data.compaction_threshold,
-            request_data.gc_horizon,
-            &request_data.gc_period,
-            request_data.image_creation_threshold,
-            &request_data.pitr_interval,
-            &request_data.walreceiver_connect_timeout,
-            &request_data.lagging_wal_timeout,
-            request_data.max_lsn_wal_lag,
-            request_data.trace_read_requests,
-            &request_data.eviction_policy,
-            request_data.min_resident_size_override,
-            &request_data.evictions_low_residence_duration_metric_threshold,
-        )
-    }
-}
-
 impl TryFrom<&'_ TenantConfigRequest> for TenantConfOpt {
     type Error = anyhow::Error;
 
     fn try_from(request_data: &TenantConfigRequest) -> Result<Self, Self::Error> {
-        Self::from_request(
-            request_data.checkpoint_distance,
-            &request_data.checkpoint_timeout,
-            request_data.compaction_target_size,
-            &request_data.compaction_period,
-            request_data.compaction_threshold,
-            request_data.gc_horizon,
-            &request_data.gc_period,
-            request_data.image_creation_threshold,
-            &request_data.pitr_interval,
-            &request_data.walreceiver_connect_timeout,
-            &request_data.lagging_wal_timeout,
-            request_data.max_lsn_wal_lag,
-            request_data.trace_read_requests,
-            &request_data.eviction_policy,
-            request_data.min_resident_size_override,
-            &request_data.evictions_low_residence_duration_metric_threshold,
-        )
+        let mut tenant_conf = TenantConfOpt::default();
+        if let Some(gc_period) = &request_data.gc_period {
+            tenant_conf.gc_period = Some(
+                humantime::parse_duration(gc_period)
+                    .with_context(bad_duration("gc_period", gc_period))?,
+            );
+        }
+        tenant_conf.gc_horizon = request_data.gc_horizon;
+        tenant_conf.image_creation_threshold = request_data.image_creation_threshold;
+
+        if let Some(pitr_interval) = &request_data.pitr_interval {
+            tenant_conf.pitr_interval = Some(
+                humantime::parse_duration(pitr_interval)
+                    .with_context(bad_duration("pitr_interval", pitr_interval))?,
+            );
+        }
+        if let Some(walreceiver_connect_timeout) = &request_data.walreceiver_connect_timeout {
+            tenant_conf.walreceiver_connect_timeout = Some(
+                humantime::parse_duration(walreceiver_connect_timeout).with_context(
+                    bad_duration("walreceiver_connect_timeout", walreceiver_connect_timeout),
+                )?,
+            );
+        }
+        if let Some(lagging_wal_timeout) = &request_data.lagging_wal_timeout {
+            tenant_conf.lagging_wal_timeout = Some(
+                humantime::parse_duration(lagging_wal_timeout)
+                    .with_context(bad_duration("lagging_wal_timeout", lagging_wal_timeout))?,
+            );
+        }
+        tenant_conf.max_lsn_wal_lag = request_data.max_lsn_wal_lag;
+        tenant_conf.trace_read_requests = request_data.trace_read_requests;
+
+        tenant_conf.checkpoint_distance = request_data.checkpoint_distance;
+        if let Some(checkpoint_timeout) = &request_data.checkpoint_timeout {
+            tenant_conf.checkpoint_timeout = Some(
+                humantime::parse_duration(checkpoint_timeout)
+                    .with_context(bad_duration("checkpoint_timeout", checkpoint_timeout))?,
+            );
+        }
+        tenant_conf.compaction_target_size = request_data.compaction_target_size;
+        tenant_conf.compaction_threshold = request_data.compaction_threshold;
+
+        if let Some(compaction_period) = &request_data.compaction_period {
+            tenant_conf.compaction_period = Some(
+                humantime::parse_duration(compaction_period)
+                    .with_context(bad_duration("compaction_period", compaction_period))?,
+            );
+        }
+
+        if let Some(eviction_policy) = &request_data.eviction_policy {
+            tenant_conf.eviction_policy = Some(
+                serde::Deserialize::deserialize(eviction_policy)
+                    .context("parse field `eviction_policy`")?,
+            );
+        }
+
+        tenant_conf.min_resident_size_override = request_data.min_resident_size_override;
+
+        if let Some(evictions_low_residence_duration_metric_threshold) =
+            &request_data.evictions_low_residence_duration_metric_threshold
+        {
+            tenant_conf.evictions_low_residence_duration_metric_threshold = Some(
+                humantime::parse_duration(evictions_low_residence_duration_metric_threshold)
+                    .with_context(bad_duration(
+                        "evictions_low_residence_duration_metric_threshold",
+                        evictions_low_residence_duration_metric_threshold,
+                    ))?,
+            );
+        }
+
+        Ok(tenant_conf)
     }
 }
 

pageserver/src/tenant/layer_map.rs

@@ -285,11 +285,6 @@ where
         }
     }
 
-    pub fn contains(&self, key_range: &Range<Key>, lsn_range: &Range<Lsn>, is_image: bool) -> bool {
-        let key = historic_layer_coverage::LayerKey::from_ranges(key_range, lsn_range, is_image);
-        self.historic.contains(&key)
-    }
-
     ///
     /// Remove an on-disk layer from the map.
     ///

pageserver/src/tenant/layer_map/historic_layer_coverage.rs

@@ -53,20 +53,6 @@ impl<'a, L: crate::tenant::storage_layer::Layer + ?Sized> From<&'a L> for LayerK
     }
 }
 
-impl LayerKey {
-    pub fn from_ranges(
-        kr: &Range<crate::tenant::layer_map::Key>,
-        lr: &Range<utils::lsn::Lsn>,
-        is_image: bool,
-    ) -> Self {
-        LayerKey {
-            key: kr.start.to_i128()..kr.end.to_i128(),
-            lsn: lr.start.0..lr.end.0,
-            is_image,
-        }
-    }
-}
-
 /// Efficiently queryable layer coverage for each LSN.
 ///
 /// Allows answering layer map queries very efficiently,
@@ -431,14 +417,6 @@ impl<Value: Clone> BufferedHistoricLayerCoverage<Value> {
         }
     }
 
-    pub fn contains(&self, layer_key: &LayerKey) -> bool {
-        match self.buffer.get(layer_key) {
-            Some(None) => false,                         // layer remove was buffered
-            Some(_) => true,                             // layer insert was buffered
-            None => self.layers.contains_key(layer_key), // no buffered ops for this layer
-        }
-    }
-
     pub fn insert(&mut self, layer_key: LayerKey, value: Value) {
         self.buffer.insert(layer_key, Some(value));
     }

pageserver/src/tenant/mgr.rs

@@ -186,20 +186,10 @@ pub fn schedule_local_tenant_processing(
     let tenant = if conf.tenant_attaching_mark_file_path(&tenant_id).exists() {
         info!("tenant {tenant_id} has attaching mark file, resuming its attach operation");
         if let Some(remote_storage) = remote_storage {
-            match Tenant::spawn_attach(conf, tenant_id, remote_storage, ctx) {
-                Ok(tenant) => tenant,
-                Err(e) => {
-                    error!("Failed to spawn_attach tenant {tenant_id}, reason: {e:#}");
-                    Tenant::create_broken_tenant(conf, tenant_id, format!("{e:#}"))
-                }
-            }
+            Tenant::spawn_attach(conf, tenant_id, remote_storage, ctx)
         } else {
             warn!("tenant {tenant_id} has attaching mark file, but pageserver has no remote storage configured");
-            Tenant::create_broken_tenant(
-                conf,
-                tenant_id,
-                "attaching mark file present but no remote storage configured".to_string(),
-            )
+            Tenant::create_broken_tenant(conf, tenant_id)
         }
     } else {
         info!("tenant {tenant_id} is assumed to be loadable, starting load operation");
@@ -476,8 +466,7 @@ pub async fn attach_tenant(
             "Cannot attach tenant {tenant_id}, local tenant directory already exists"
         );
 
-        let tenant =
-            Tenant::spawn_attach(conf, tenant_id, remote_storage, ctx).context("spawn_attach")?;
+        let tenant = Tenant::spawn_attach(conf, tenant_id, remote_storage, ctx);
         vacant_entry.insert(tenant);
         Ok(())
     })

pageserver/src/tenant/size.rs

@@ -11,7 +11,7 @@ use tokio_util::sync::CancellationToken;
 use crate::context::RequestContext;
 use crate::pgdatadir_mapping::CalculateLogicalSizeError;
 
-use super::{LogicalSizeCalculationCause, Tenant};
+use super::Tenant;
 use crate::tenant::Timeline;
 use utils::id::TimelineId;
 use utils::lsn::Lsn;
@@ -126,7 +126,6 @@ pub(super) async fn gather_inputs(
     limit: &Arc<Semaphore>,
     max_retention_period: Option<u64>,
     logical_size_cache: &mut HashMap<(TimelineId, Lsn), u64>,
-    cause: LogicalSizeCalculationCause,
     ctx: &RequestContext,
 ) -> anyhow::Result<ModelInputs> {
     // refresh is needed to update gc related pitr_cutoff and horizon_cutoff
@@ -319,15 +318,7 @@ pub(super) async fn gather_inputs(
 
     // We left the 'size' field empty in all of the Segments so far.
     // Now find logical sizes for all of the points that might need or benefit from them.
-    fill_logical_sizes(
-        &timelines,
-        &mut segments,
-        limit,
-        logical_size_cache,
-        cause,
-        ctx,
-    )
-    .await?;
+    fill_logical_sizes(&timelines, &mut segments, limit, logical_size_cache, ctx).await?;
 
     Ok(ModelInputs {
         segments,
@@ -345,7 +336,6 @@ async fn fill_logical_sizes(
     segments: &mut [SegmentMeta],
     limit: &Arc<Semaphore>,
     logical_size_cache: &mut HashMap<(TimelineId, Lsn), u64>,
-    cause: LogicalSizeCalculationCause,
     ctx: &RequestContext,
 ) -> anyhow::Result<()> {
     let timeline_hash: HashMap<TimelineId, Arc<Timeline>> = HashMap::from_iter(
@@ -383,17 +373,13 @@ async fn fill_logical_sizes(
                 let timeline = Arc::clone(timeline_hash.get(&timeline_id).unwrap());
                 let parallel_size_calcs = Arc::clone(limit);
                 let ctx = ctx.attached_child();
-                joinset.spawn(
-                    calculate_logical_size(
-                        parallel_size_calcs,
-                        timeline,
-                        lsn,
-                        cause,
-                        ctx,
-                        cancel.child_token(),
-                    )
-                    .in_current_span(),
-                );
+                joinset.spawn(calculate_logical_size(
+                    parallel_size_calcs,
+                    timeline,
+                    lsn,
+                    ctx,
+                    cancel.child_token(),
+                ));
             }
             e.insert(cached_size);
         }
@@ -496,7 +482,6 @@ async fn calculate_logical_size(
     limit: Arc<tokio::sync::Semaphore>,
     timeline: Arc<crate::tenant::Timeline>,
     lsn: utils::lsn::Lsn,
-    cause: LogicalSizeCalculationCause,
     ctx: RequestContext,
     cancel: CancellationToken,
 ) -> Result<TimelineAtLsnSizeResult, RecvError> {
@@ -505,7 +490,7 @@ async fn calculate_logical_size(
         .expect("global semaphore should not had been closed");
 
     let size_res = timeline
-        .spawn_ondemand_logical_size_calculation(lsn, cause, ctx, cancel)
+        .spawn_ondemand_logical_size_calculation(lsn, ctx, cancel)
         .instrument(info_span!("spawn_ondemand_logical_size_calculation"))
         .await?;
     Ok(TimelineAtLsnSizeResult(timeline, lsn, size_res))

pageserver/src/tenant/storage_layer/delta_layer.rs

@@ -867,14 +867,6 @@ pub struct DeltaLayerWriter {
 }
 
 impl DeltaLayerWriter {
-    pub fn key_start(&self) -> Key {
-        self.inner.as_ref().unwrap().key_start
-    }
-
-    pub fn lsn_range(&self) -> Range<Lsn> {
-        self.inner.as_ref().unwrap().lsn_range.clone()
-    }
-
     ///
     /// Start building a new delta layer.
     ///

pageserver/src/tenant/tasks.rs

@@ -259,7 +259,6 @@ pub(crate) async fn random_init_delay(
     }
 }
 
-/// Attention: the `task` and `period` beocme labels of a pageserver-wide prometheus metric.
 pub(crate) fn warn_when_period_overrun(elapsed: Duration, period: Duration, task: &str) {
     // Duration::ZERO will happen because it's the "disable [bgtask]" value.
     if elapsed >= period && period != Duration::ZERO {
@@ -272,8 +271,5 @@ pub(crate) fn warn_when_period_overrun(elapsed: Duration, period: Duration, task
             task,
             "task iteration took longer than the configured period"
         );
-        crate::metrics::BACKGROUND_LOOP_PERIOD_OVERRUN_COUNT
-            .with_label_values(&[task, &format!("{}", period.as_secs())])
-            .inc();
     }
 }

pageserver/src/tenant/timeline.rs

@@ -227,6 +227,9 @@ pub struct Timeline {
     state: watch::Sender<TimelineState>,
 
     eviction_task_timeline_state: tokio::sync::Mutex<EvictionTaskTimelineState>,
+
+    pub(super) delete_self:
+        utils::shared_retryable::SharedRetryable<Result<(), super::InnerDeleteTimelineError>>,
 }
 
 /// Internal structure to hold all data needed for logical size calculation.
@@ -396,9 +399,6 @@ pub enum PageReconstructError {
     /// The operation was cancelled
     Cancelled,
 
-    /// The ancestor of this is being stopped
-    AncestorStopping(TimelineId),
-
     /// An error happened replaying WAL records
     #[error(transparent)]
     WalRedo(#[from] crate::walredo::WalRedoError),
@@ -417,9 +417,6 @@ impl std::fmt::Debug for PageReconstructError {
                 )
             }
             Self::Cancelled => write!(f, "cancelled"),
-            Self::AncestorStopping(timeline_id) => {
-                write!(f, "ancestor timeline {timeline_id} is being stopped")
-            }
             Self::WalRedo(err) => err.fmt(f),
         }
     }
@@ -438,22 +435,11 @@ impl std::fmt::Display for PageReconstructError {
                 )
             }
             Self::Cancelled => write!(f, "cancelled"),
-            Self::AncestorStopping(timeline_id) => {
-                write!(f, "ancestor timeline {timeline_id} is being stopped")
-            }
             Self::WalRedo(err) => err.fmt(f),
         }
     }
 }
 
-#[derive(Clone, Copy)]
-pub enum LogicalSizeCalculationCause {
-    Initial,
-    ConsumptionMetricsSyntheticSize,
-    EvictionTaskImitation,
-    TenantSizeHandler,
-}
-
 /// Public interface functions
 impl Timeline {
     /// Get the LSN where this branch was created
@@ -943,31 +929,6 @@ impl Timeline {
         self.state.subscribe()
     }
 
-    pub async fn wait_to_become_active(
-        &self,
-        _ctx: &RequestContext, /* Prepare for use by cancellation */
-    ) -> Result<(), TimelineState> {
-        let mut receiver = self.state.subscribe();
-        loop {
-            let current_state = *receiver.borrow_and_update();
-            match current_state {
-                TimelineState::Loading => {
-                    receiver
-                        .changed()
-                        .await
-                        .expect("holding a reference to self");
-                }
-                TimelineState::Active { .. } => {
-                    return Ok(());
-                }
-                TimelineState::Broken { .. } | TimelineState::Stopping => {
-                    // There's no chance the timeline can transition back into ::Active
-                    return Err(current_state);
-                }
-            }
-        }
-    }
-
     pub fn layer_map_info(&self, reset: LayerAccessStatsReset) -> LayerMapInfo {
         let layer_map = self.layers.read().unwrap();
         let mut in_memory_layers = Vec::with_capacity(layer_map.frozen_layers.len() + 1);
@@ -1421,6 +1382,8 @@ impl Timeline {
                 eviction_task_timeline_state: tokio::sync::Mutex::new(
                     EvictionTaskTimelineState::default(),
                 ),
+
+                delete_self: utils::shared_retryable::SharedRetryable::default(),
             };
             result.repartition_threshold = result.get_checkpoint_distance() / 10;
             result
@@ -1881,31 +1844,18 @@ impl Timeline {
                 // to spawn_ondemand_logical_size_calculation.
                 let cancel = CancellationToken::new();
                 let calculated_size = match self_clone
-                    .logical_size_calculation_task(lsn, LogicalSizeCalculationCause::Initial, &background_ctx, cancel)
+                    .logical_size_calculation_task(lsn, &background_ctx, cancel)
                     .await
                 {
                     Ok(s) => s,
                     Err(CalculateLogicalSizeError::Cancelled) => {
                         // Don't make noise, this is a common task.
-                        // In the unlikely case that there is another call to this function, we'll retry
+                        // In the unlikely case that there ihs another call to this function, we'll retry
                         // because initial_logical_size is still None.
                         info!("initial size calculation cancelled, likely timeline delete / tenant detach");
                         return Ok(());
                     }
-                    Err(CalculateLogicalSizeError::Other(err)) => {
-                        if let Some(e @ PageReconstructError::AncestorStopping(_)) =
-                            err.root_cause().downcast_ref()
-                        {
-                            // This can happen if the timeline parent timeline switches to
-                            // Stopping state while we're still calculating the initial
-                            // timeline size for the child, for example if the tenant is
-                            // being detached or the pageserver is shut down. Like with
-                            // CalculateLogicalSizeError::Cancelled, don't make noise.
-                            info!("initial size calculation failed because the timeline or its ancestor is Stopping, likely because the tenant is being detached: {e:#}");
-                            return Ok(());
-                        }
-                        return Err(err.context("Failed to calculate logical size"));
-                    }
+                    x @ Err(_) => x.context("Failed to calculate logical size")?,
                 };
 
                 // we cannot query current_logical_size.current_size() to know the current
@@ -1941,14 +1891,13 @@ impl Timeline {
                 // so that we prevent future callers from spawning this task
                 permit.forget();
                 Ok(())
-            }.in_current_span(),
+            },
         );
     }
 
     pub fn spawn_ondemand_logical_size_calculation(
         self: &Arc<Self>,
         lsn: Lsn,
-        cause: LogicalSizeCalculationCause,
         ctx: RequestContext,
         cancel: CancellationToken,
     ) -> oneshot::Receiver<Result<u64, CalculateLogicalSizeError>> {
@@ -1971,26 +1920,22 @@ impl Timeline {
             false,
             async move {
                 let res = self_clone
-                    .logical_size_calculation_task(lsn, cause, &ctx, cancel)
+                    .logical_size_calculation_task(lsn, &ctx, cancel)
                     .await;
                 let _ = sender.send(res).ok();
                 Ok(()) // Receiver is responsible for handling errors
-            }
-            .in_current_span(),
+            },
         );
         receiver
     }
 
-    #[instrument(skip_all)]
+    #[instrument(skip_all, fields(tenant = %self.tenant_id, timeline = %self.timeline_id))]
     async fn logical_size_calculation_task(
         self: &Arc<Self>,
         lsn: Lsn,
-        cause: LogicalSizeCalculationCause,
         ctx: &RequestContext,
         cancel: CancellationToken,
     ) -> Result<u64, CalculateLogicalSizeError> {
-        debug_assert_current_span_has_tenant_and_timeline_id();
-
         let mut timeline_state_updates = self.subscribe_for_state_updates();
         let self_calculation = Arc::clone(self);
 
@@ -1998,7 +1943,7 @@ impl Timeline {
             let cancel = cancel.child_token();
             let ctx = ctx.attached_child();
             self_calculation
-                .calculate_logical_size(lsn, cause, cancel, &ctx)
+                .calculate_logical_size(lsn, cancel, &ctx)
                 .await
         });
         let timeline_state_cancellation = async {
@@ -2053,7 +1998,6 @@ impl Timeline {
     pub async fn calculate_logical_size(
         &self,
         up_to_lsn: Lsn,
-        cause: LogicalSizeCalculationCause,
         cancel: CancellationToken,
         ctx: &RequestContext,
     ) -> Result<u64, CalculateLogicalSizeError> {
@@ -2087,15 +2031,7 @@ impl Timeline {
         if let Some(size) = self.current_logical_size.initialized_size(up_to_lsn) {
             return Ok(size);
         }
-        let storage_time_metrics = match cause {
-            LogicalSizeCalculationCause::Initial
-            | LogicalSizeCalculationCause::ConsumptionMetricsSyntheticSize
-            | LogicalSizeCalculationCause::TenantSizeHandler => &self.metrics.logical_size_histo,
-            LogicalSizeCalculationCause::EvictionTaskImitation => {
-                &self.metrics.imitate_logical_size_histo
-            }
-        };
-        let timer = storage_time_metrics.start_timer();
+        let timer = self.metrics.logical_size_histo.start_timer();
         let logical_size = self
             .get_current_logical_size_non_incremental(up_to_lsn, cancel, ctx)
             .await?;
@@ -2287,46 +2223,6 @@ impl Timeline {
                     Ok(timeline) => timeline,
                     Err(e) => return Err(PageReconstructError::from(e)),
                 };
-
-                // It's possible that the ancestor timeline isn't active yet, or
-                // is active but hasn't yet caught up to the branch point. Wait
-                // for it.
-                //
-                // This cannot happen while the pageserver is running normally,
-                // because you cannot create a branch from a point that isn't
-                // present in the pageserver yet. However, we don't wait for the
-                // branch point to be uploaded to cloud storage before creating
-                // a branch. I.e., the branch LSN need not be remote consistent
-                // for the branching operation to succeed.
-                //
-                // Hence, if we try to load a tenant in such a state where
-                // 1. the existence of the branch was persisted (in IndexPart and/or locally)
-                // 2. but the ancestor state is behind branch_lsn because it was not yet persisted
-                // then we will need to wait for the ancestor timeline to
-                // re-stream WAL up to branch_lsn before we access it.
-                //
-                // How can a tenant get in such a state?
-                // - ungraceful pageserver process exit
-                // - detach+attach => this is a bug, https://github.com/neondatabase/neon/issues/4219
-                //
-                // NB: this could be avoided by requiring
-                //   branch_lsn >= remote_consistent_lsn
-                // during branch creation.
-                match ancestor.wait_to_become_active(ctx).await {
-                    Ok(()) => {}
-                    Err(state) if state == TimelineState::Stopping => {
-                        return Err(PageReconstructError::AncestorStopping(ancestor.timeline_id));
-                    }
-                    Err(state) => {
-                        return Err(PageReconstructError::Other(anyhow::anyhow!(
-                            "Timeline {} will not become active. Current state: {:?}",
-                            ancestor.timeline_id,
-                            &state,
-                        )));
-                    }
-                }
-                ancestor.wait_lsn(timeline.ancestor_lsn, ctx).await?;
-
                 timeline_owned = ancestor;
                 timeline = &*timeline_owned;
                 prev_lsn = Lsn(u64::MAX);
@@ -3375,54 +3271,7 @@ impl Timeline {
                         || contains_hole
                     {
                         // ... if so, flush previous layer and prepare to write new one
-
-                        let end_key = prev_key.unwrap().next();
-
-                        let w = writer.take().unwrap();
-
-                        // If an identical L1 layer already exists, no need to create a new one.
-                        //
-                        // This can happen if compaction is interrupted after it has already
-                        // created some or all of the L1 layers, but has not deleted the L0 layers
-                        // yet, so that on next compaction, we do the same work again.
-                        //
-                        // NOTE: this is racy, if there can be any other task that concurrently
-                        // creates L1 layers. Currently, there can be only one compaction task
-                        // running at any time, so this is fine.
-                        //
-                        // Also we hold `layer_removal_cs` guard which should prevent race condition
-                        // even if there are two or more concurrent compaction tasks.
-                        //
-                        // But there is an opposite issue: we check presence of duplicates under
-                        // `layers` shared lock, but then it is released. So there is a gap between
-                        // this check and adding new layer to layer map. In principle in this gap some
-                        // some other task (i.e. GC) can drop this layer and we already abandon insertion
-                        // of duplicate layer. As a result there will be no such layer at all.
-                        // In other words: we have some state S1 of pageserver where layer L1 can be removed by GC.
-                        // Then we run compaction and it switch pageserver to the state S2 which writes duplicate of
-                        // layer L1 and where it can not be removed. With this patch it is possible that
-                        // we switch pageserver to state S2 but... with L1 lost.
-                        // It is just hypothetical situation and there is no such concrete scenario which
-                        // reproduces this problem. So let's take this risk.
-                        //
-                        if self.layers.read().unwrap().contains(
-                            &(w.key_start()..end_key),
-                            &w.lsn_range(),
-                            false, // not an image layer
-                        ) {
-                            info!(
-                                "Skip generation of duplicate layer {}_{}__{}_{}",
-                                w.key_start(),
-                                end_key,
-                                w.lsn_range().start,
-                                w.lsn_range().end
-                            );
-                            drop(w);
-                        } else {
-                            let new_layer = w.finish(end_key)?;
-
-                            new_layers.push(new_layer);
-                        }
+                        new_layers.push(writer.take().unwrap().finish(prev_key.unwrap().next())?);
                         writer = None;
 
                         if contains_hole {
@@ -3479,10 +3328,6 @@ impl Timeline {
 
         drop(all_keys_iter); // So that deltas_to_compact is no longer borrowed
 
-        fail_point!("compact-level0-phase1-finish", |_| {
-            Err(anyhow::anyhow!("failpoint compact-level0-phase1-finish").into())
-        });
-
         Ok(CompactLevel0Phase1Result {
             new_layers,
             deltas_to_compact,

pageserver/src/tenant/timeline/eviction_task.rs

@@ -30,7 +30,7 @@ use crate::{
     tenant::{
         config::{EvictionPolicy, EvictionPolicyLayerAccessThreshold},
         storage_layer::PersistentLayer,
-        LogicalSizeCalculationCause, Tenant,
+        Tenant,
     },
 };
 
@@ -120,13 +120,6 @@ impl Timeline {
                 }
                 let elapsed = start.elapsed();
                 crate::tenant::tasks::warn_when_period_overrun(elapsed, p.period, "eviction");
-                crate::metrics::EVICTION_ITERATION_DURATION
-                    .get_metric_with_label_values(&[
-                        &format!("{}", p.period.as_secs()),
-                        &format!("{}", p.threshold.as_secs()),
-                    ])
-                    .unwrap()
-                    .observe(elapsed.as_secs_f64());
                 ControlFlow::Continue(start + p.period)
             }
         }
@@ -342,12 +335,7 @@ impl Timeline {
 
         // imitiate on-restart initial logical size
         let size = self
-            .calculate_logical_size(
-                lsn,
-                LogicalSizeCalculationCause::EvictionTaskImitation,
-                cancel.clone(),
-                ctx,
-            )
+            .calculate_logical_size(lsn, cancel.clone(), ctx)
             .instrument(info_span!("calculate_logical_size"))
             .await;
 
@@ -419,15 +407,9 @@ impl Timeline {
             .inner();
 
         let mut throwaway_cache = HashMap::new();
-        let gather = crate::tenant::size::gather_inputs(
-            tenant,
-            limit,
-            None,
-            &mut throwaway_cache,
-            LogicalSizeCalculationCause::EvictionTaskImitation,
-            ctx,
-        )
-        .instrument(info_span!("gather_inputs"));
+        let gather =
+            crate::tenant::size::gather_inputs(tenant, limit, None, &mut throwaway_cache, ctx)
+                .instrument(info_span!("gather_inputs"));
 
         tokio::select! {
             _ = cancel.cancelled() => {}

pageserver/src/walingest.rs

@@ -305,15 +305,6 @@ impl<'a> WalIngest<'a> {
                     self.checkpoint_modified = true;
                 }
             }
-        } else if decoded.xl_rmid == pg_constants::RM_LOGICALMSG_ID {
-            let info = decoded.xl_info & pg_constants::XLR_RMGR_INFO_MASK;
-            if info == pg_constants::XLOG_LOGICAL_MESSAGE {
-                // This is a convenient way to make the WAL ingestion pause at
-                // particular point in the WAL. For more fine-grained control,
-                // we could peek into the message and only pause if it contains
-                // a particular string, for example, but this is enough for now.
-                utils::failpoint_sleep_millis_async!("wal-ingest-logical-message-sleep");
-            }
         }
 
         // Iterate through all the blocks that the record modifies, and

pgxn/neon/pagestore_client.h

@@ -52,7 +52,7 @@ typedef struct
 #define NEON_TAG "[NEON_SMGR] "
 #define neon_log(tag, fmt, ...) ereport(tag,                                  \
 										(errmsg(NEON_TAG fmt, ##__VA_ARGS__), \
-										 errhidestmt(true), errhidecontext(true), errposition(0), internalerrposition(0)))
+										 errhidestmt(true), errhidecontext(true), internalerrposition(0)))
 
 /*
  * supertype of all the Neon*Request structs below

scripts/flaky_tests.py

@@ -21,7 +21,6 @@ FLAKY_TESTS_QUERY = """
                 jsonb_array_elements(jsonb_array_elements(data -> 'children') -> 'children') -> 'name' as suite,
                 jsonb_array_elements(jsonb_array_elements(jsonb_array_elements(data -> 'children') -> 'children') -> 'children') -> 'name' as test,
                 jsonb_array_elements(jsonb_array_elements(jsonb_array_elements(data -> 'children') -> 'children') -> 'children') -> 'status' as status,
-                jsonb_array_elements(jsonb_array_elements(jsonb_array_elements(data -> 'children') -> 'children') -> 'children') -> 'retriesStatusChange' as retries_status_change,
                 to_timestamp((jsonb_array_elements(jsonb_array_elements(jsonb_array_elements(data -> 'children') -> 'children') -> 'children') -> 'time' -> 'start')::bigint / 1000)::date as timestamp
             FROM
                 regress_test_results
@@ -30,7 +29,7 @@ FLAKY_TESTS_QUERY = """
         ) data
     WHERE
         timestamp > CURRENT_DATE - INTERVAL '%s' day
-        AND (status::text IN ('"failed"', '"broken"') OR retries_status_change::boolean)
+        AND status::text IN ('"failed"', '"broken"')
     ;
 """
 

scripts/pr-comment-test-report.js

@@ -1,5 +1,6 @@
 //
 // The script parses Allure reports and posts a comment with a summary of the test results to the PR.
+// It accepts an array of items and creates a comment with a summary for each one (for "release" and "debug", together or separately if any of them failed to be generated).
 //
 // The comment is updated on each run with the latest results.
 //
@@ -12,37 +13,19 @@
 //         github,
 //         context,
 //         fetch,
-//         report: {
-//           reportUrl: "...",
-//           reportJsonUrl: "...",
-//         },
+//         reports: [{...}, ...], // each report is expected to have "buildType", "reportUrl", and "jsonUrl" properties
 //       })
 //
 
-// Analog of Python's defaultdict.
-//
-// const dm = new DefaultMap(() => new DefaultMap(() => []))
-// dm["firstKey"]["secondKey"].push("value")
-//
-class DefaultMap extends Map {
-    constructor(getDefaultValue) {
-        return new Proxy({}, {
-            get: (target, name) => name in target ? target[name] : (target[name] = getDefaultValue(name))
-        })
-    }
-}
-
-module.exports = async ({ github, context, fetch, report }) => {
+module.exports = async ({ github, context, fetch, reports }) => {
     // Marker to find the comment in the subsequent runs
     const startMarker = `<!--AUTOMATIC COMMENT START #${context.payload.number}-->`
-    // Let users know that the comment is updated automatically
-    const autoupdateNotice = `<div align="right"><sub>The comment gets automatically updated with the latest test results :recycle:</sub></div>`
     // GitHub bot id taken from (https://api.github.com/users/github-actions[bot])
     const githubActionsBotId = 41898282
     // The latest commit in the PR URL
     const commitUrl = `${context.serverUrl}/${context.repo.owner}/${context.repo.repo}/pull/${context.payload.number}/commits/${context.payload.pull_request.head.sha}`
     // Commend body itself
-    let commentBody = `${startMarker}\n`
+    let commentBody = `${startMarker}\n### Test results for ${commitUrl}:\n___\n`
 
     // Common parameters for GitHub API requests
     const ownerRepoParams = {
@@ -50,119 +33,76 @@ module.exports = async ({ github, context, fetch, report }) => {
         repo: context.repo.repo,
     }
 
-    const {reportUrl, reportJsonUrl} = report
+    for (const report of reports) {
+        const {buildType, reportUrl, jsonUrl} = report
 
-    if (!reportUrl || !reportJsonUrl) {
-        commentBody += `#### No tests were run or test report is not available\n`
-        commentBody += autoupdateNotice
-        return
-    }
+        if (!reportUrl || !jsonUrl) {
+            commentBody += `#### ${buildType} build: no tests were run or test report is not available\n`
+            continue
+        }
 
-    const suites = await (await fetch(reportJsonUrl)).json()
+        const suites = await (await fetch(jsonUrl)).json()
 
-    // Allure distinguishes "failed" (with an assertion error) and "broken" (with any other error) tests.
-    // For this report it's ok to treat them in the same way (as failed).
-    const failedTests = new DefaultMap(() => new DefaultMap(() => []))
-    const passedTests = new DefaultMap(() => new DefaultMap(() => []))
-    const skippedTests = new DefaultMap(() => new DefaultMap(() => []))
-    const retriedTests = new DefaultMap(() => new DefaultMap(() => []))
-    const flakyTests = new DefaultMap(() => new DefaultMap(() => []))
+        // Allure distinguishes "failed" (with an assertion error) and "broken" (with any other error) tests.
+        // For this report it's ok to treat them in the same way (as failed).
+        failedTests = []
+        passedTests = []
+        skippedTests = []
 
-    let failedTestsCount = 0
-    let passedTestsCount = 0
-    let skippedTestsCount = 0
-    let flakyTestsCount = 0
+        retriedTests = []
+        retriedStatusChangedTests = []
 
-    const pgVersions = new Set()
-    const buildTypes = new Set()
+        for (const parentSuite of suites.children) {
+            for (const suite of parentSuite.children) {
+                for (const test of suite.children) {
+                    pytestName = `${parentSuite.name.replace(".", "/")}/${suite.name}.py::${test.name}`
+                    test.pytestName = pytestName
 
-    for (const parentSuite of suites.children) {
-        for (const suite of parentSuite.children) {
-            for (const test of suite.children) {
-                let buildType, pgVersion
-                const match = test.name.match(/[\[-](?<buildType>debug|release)-pg(?<pgVersion>\d+)[-\]]/)?.groups
-                if (match) {
-                    ({buildType, pgVersion} = match)
-                } else {
-                    // It's ok, we embed BUILD_TYPE and Postgres Version into the test name only for regress suite and do not for other suites (like performance).
-                    console.info(`Cannot get BUILD_TYPE and Postgres Version from test name: "${test.name}", defaulting to "release" and "14"`)
+                    if (test.status === "passed") {
+                        passedTests.push(test);
+                    } else if (test.status === "failed" || test.status === "broken") {
+                        failedTests.push(test);
+                    } else if (test.status === "skipped") {
+                        skippedTests.push(test);
+                    }
 
-                    buildType = "release"
-                    pgVersion = "14"
-                }
+                    if (test.retriesCount > 0) {
+                        retriedTests.push(test);
 
-                pgVersions.add(pgVersion)
-                buildTypes.add(buildType)
-
-                // Removing build type and PostgreSQL version from the test name to make it shorter
-                const testName = test.name.replace(new RegExp(`${buildType}-pg${pgVersion}-?`), "").replace("[]", "")
-                test.pytestName = `${parentSuite.name.replace(".", "/")}/${suite.name}.py::${testName}`
-
-                if (test.status === "passed") {
-                    passedTests[pgVersion][buildType].push(test)
-                    passedTestsCount += 1
-                } else if (test.status === "failed" || test.status === "broken") {
-                    failedTests[pgVersion][buildType].push(test)
-                    failedTestsCount += 1
-                } else if (test.status === "skipped") {
-                    skippedTests[pgVersion][buildType].push(test)
-                    skippedTestsCount += 1
-                }
-
-                if (test.retriesCount > 0) {
-                    retriedTests[pgVersion][buildType].push(test)
-
-                    if (test.retriesStatusChange) {
-                        flakyTests[pgVersion][buildType].push(test)
-                        flakyTestsCount += 1
+                        if (test.retriedStatusChangedTests) {
+                            retriedStatusChangedTests.push(test);
+                        }
                     }
                 }
             }
         }
-    }
 
-    const totalTestsCount = failedTestsCount + passedTestsCount + skippedTestsCount
-    commentBody += `### ${totalTestsCount} tests run: ${passedTestsCount} passed, ${failedTestsCount} failed, ${skippedTestsCount} skipped ([full report](${reportUrl}) for ${commitUrl})\n___\n`
+        const totalTestsCount = failedTests.length + passedTests.length + skippedTests.length
+        commentBody += `#### ${buildType} build: ${totalTestsCount} tests run: ${passedTests.length} passed, ${failedTests.length} failed, ${skippedTests.length} skipped ([full report](${reportUrl}))\n`
+        if (failedTests.length > 0) {
+            commentBody += `Failed tests:\n`
+            for (const test of failedTests) {
+                const allureLink = `${reportUrl}#suites/${test.parentUid}/${test.uid}`
 
-    // Print test resuls from the newest to the oldest PostgreSQL version for release and debug builds.
-    for (const pgVersion of Array.from(pgVersions).sort().reverse()) {
-        for (const buildType of Array.from(buildTypes).sort().reverse()) {
-            if (failedTests[pgVersion][buildType].length > 0) {
-                commentBody += `#### PostgreSQL ${pgVersion} (${buildType} build)\n\n`
-                commentBody += `Failed tests:\n`
-                for (const test of failedTests[pgVersion][buildType]) {
-                    const allureLink = `${reportUrl}#suites/${test.parentUid}/${test.uid}`
-
-                    commentBody += `- [\`${test.pytestName}\`](${allureLink})`
-                    if (test.retriesCount > 0) {
-                        commentBody += ` (ran [${test.retriesCount + 1} times](${allureLink}/retries))`
-                    }
-                    commentBody += "\n"
+                commentBody += `- [\`${test.pytestName}\`](${allureLink})`
+                if (test.retriesCount > 0) {
+                    commentBody += ` (ran [${test.retriesCount + 1} times](${allureLink}/retries))`
                 }
                 commentBody += "\n"
             }
+            commentBody += "\n"
         }
-    }
-
-    if (flakyTestsCount > 0) {
-        commentBody += "<details>\n<summary>Flaky tests</summary>\n\n"
-        for (const pgVersion of Array.from(pgVersions).sort().reverse()) {
-            for (const buildType of Array.from(buildTypes).sort().reverse()) {
-                if (flakyTests[pgVersion][buildType].length > 0) {
-                    commentBody += `#### PostgreSQL ${pgVersion} (${buildType} build)\n\n`
-                    for (const test of flakyTests[pgVersion][buildType]) {
-                        const status = test.status === "passed" ? ":white_check_mark:" : ":x:"
-                        commentBody += `- ${status} [\`${test.pytestName}\`](${reportUrl}#suites/${test.parentUid}/${test.uid}/retries)\n`
-                    }
-                    commentBody += "\n"
-                }
+        if (retriedStatusChangedTests > 0) {
+            commentBody += `Flaky tests:\n`
+            for (const test of retriedStatusChangedTests) {
+                const status = test.status === "passed" ? ":white_check_mark:" : ":x:"
+                commentBody += `- ${status} [\`${test.pytestName}\`](${reportUrl}#suites/${test.parentUid}/${test.uid}/retries)\n`
             }
+            commentBody += "\n"
         }
-        commentBody += "\n</details>\n"
+        commentBody += "___\n"
     }
 
-    commentBody += autoupdateNotice
-
     const { data: comments } = await github.rest.issues.listComments({
         issue_number: context.payload.number,
         ...ownerRepoParams,

test_runner/conftest.py

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

test_runner/fixtures/allure.py

@@ -1,25 +0,0 @@
-import os
-
-import pytest
-
-from fixtures.pg_version import DEFAULT_VERSION, PgVersion
-
-"""
-Set of utilities to make Allure report more informative.
-
-- It adds BUILD_TYPE and DEFAULT_PG_VERSION to the test names (only in test_runner/regress)
-to make tests distinguishable in Allure report.
-"""
-
-
-@pytest.fixture(scope="function", autouse=True)
-def allure_noop():
-    pass
-
-
-def pytest_generate_tests(metafunc):
-    if "test_runner/regress" in metafunc.definition._nodeid:
-        build_type = os.environ.get("BUILD_TYPE", "DEBUG").lower()
-        pg_version = PgVersion(os.environ.get("DEFAULT_PG_VERSION", DEFAULT_VERSION))
-
-        metafunc.parametrize("allure_noop", [f"{build_type}-pg{pg_version}"])

test_runner/fixtures/benchmark_fixture.py

@@ -451,17 +451,13 @@ def pytest_terminal_summary(
     revision = os.getenv("GITHUB_SHA", "local")
     platform = os.getenv("PLATFORM", "local")
 
-    is_header_printed = False
+    terminalreporter.section("Benchmark results", "-")
 
     result = []
     for test_report in terminalreporter.stats.get("passed", []):
         result_entry = []
 
         for _, recorded_property in test_report.user_properties:
-            if not is_header_printed:
-                terminalreporter.section("Benchmark results", "-")
-                is_header_printed = True
-
             terminalreporter.write(
                 "{}.{}: ".format(test_report.head_line, recorded_property["name"])
             )
@@ -489,6 +485,7 @@ def pytest_terminal_summary(
 
     out_dir = config.getoption("out_dir")
     if out_dir is None:
+        warnings.warn("no out dir provided to store performance test results")
         return
 
     if not result:

test_runner/fixtures/pageserver/http.py

@@ -272,7 +272,6 @@ class PageserverHttpClient(requests.Session):
         new_timeline_id: Optional[TimelineId] = None,
         ancestor_timeline_id: Optional[TimelineId] = None,
         ancestor_start_lsn: Optional[Lsn] = None,
-        **kwargs,
     ) -> Dict[Any, Any]:
         body: Dict[str, Any] = {
             "new_timeline_id": str(new_timeline_id) if new_timeline_id else None,
@@ -282,9 +281,7 @@ class PageserverHttpClient(requests.Session):
         if pg_version != PgVersion.NOT_SET:
             body["pg_version"] = int(pg_version)
 
-        res = self.post(
-            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline", json=body, **kwargs
-        )
+        res = self.post(f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline", json=body)
         self.verbose_error(res)
         if res.status_code == 409:
             raise Exception(f"could not create timeline: already exists for id {new_timeline_id}")

test_runner/fixtures/pg_version.py

@@ -46,20 +46,6 @@ class PgVersion(str, enum.Enum):
 DEFAULT_VERSION: PgVersion = PgVersion.V14
 
 
-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 pytest_addoption(parser: Parser):
     parser.addoption(
         "--pg-version",

test_runner/regress/test_compatibility.py

@@ -16,7 +16,7 @@ from fixtures.neon_fixtures import (
 )
 from fixtures.pageserver.http import PageserverHttpClient
 from fixtures.pageserver.utils import wait_for_last_record_lsn, wait_for_upload
-from fixtures.pg_version import PgVersion, skip_on_postgres
+from fixtures.pg_version import PgVersion
 from fixtures.types import Lsn
 from pytest import FixtureRequest
 
@@ -41,15 +41,12 @@ check_ondisk_data_compatibility_if_enabled = pytest.mark.skipif(
 )
 
 
-@skip_on_postgres(PgVersion.V15, "Compatibility tests doesn't support Postgres 15 yet")
+# Note: if renaming this test, don't forget to update a reference to it in a workflow file:
+# "Upload compatibility snapshot" step in .github/actions/run-python-test-set/action.yml
+@check_ondisk_data_compatibility_if_enabled
 @pytest.mark.xdist_group("compatibility")
 @pytest.mark.order(before="test_forward_compatibility")
-def test_create_snapshot(
-    neon_env_builder: NeonEnvBuilder,
-    pg_bin: PgBin,
-    top_output_dir: Path,
-    test_output_dir: Path,
-):
+def test_create_snapshot(neon_env_builder: NeonEnvBuilder, pg_bin: PgBin, test_output_dir: Path):
     # The test doesn't really test anything
     # it creates a new snapshot for releases after we tested the current version against the previous snapshot in `test_backward_compatibility`.
     #
@@ -89,14 +86,10 @@ def test_create_snapshot(
         sk.stop()
     env.pageserver.stop()
 
-    # Directory `compatibility_snapshot_dir` is uploaded to S3 in a workflow, keep the name in sync with it
-    compatibility_snapshot_dir = top_output_dir / "compatibility_snapshot_pg14"
-    if compatibility_snapshot_dir.exists():
-        shutil.rmtree(compatibility_snapshot_dir)
-    shutil.copytree(test_output_dir, compatibility_snapshot_dir)
+    shutil.copytree(test_output_dir, test_output_dir / "compatibility_snapshot_pg14")
+    # Directory `test_output_dir / "compatibility_snapshot_pg14"` is uploaded to S3 in a workflow, keep the name in sync with it
 
 
-@skip_on_postgres(PgVersion.V15, "Compatibility tests doesn't support Postgres 15 yet")
 @check_ondisk_data_compatibility_if_enabled
 @pytest.mark.xdist_group("compatibility")
 @pytest.mark.order(after="test_create_snapshot")
@@ -155,13 +148,11 @@ def test_backward_compatibility(
     ), "Breaking changes are allowed by ALLOW_BACKWARD_COMPATIBILITY_BREAKAGE, but the test has passed without any breakage"
 
 
-@skip_on_postgres(PgVersion.V15, "Compatibility tests doesn't support Postgres 15 yet")
 @check_ondisk_data_compatibility_if_enabled
 @pytest.mark.xdist_group("compatibility")
 @pytest.mark.order(after="test_create_snapshot")
 def test_forward_compatibility(
     test_output_dir: Path,
-    top_output_dir: Path,
     port_distributor: PortDistributor,
     pg_version: PgVersion,
     request: FixtureRequest,
@@ -183,7 +174,9 @@ def test_forward_compatibility(
     ), "COMPATIBILITY_POSTGRES_DISTRIB_DIR is not set. It should be set to a pg_install directrory (ideally generated by the previous version of Neon)"
     compatibility_postgres_distrib_dir = Path(compatibility_postgres_distrib_dir_env).resolve()
 
-    compatibility_snapshot_dir = top_output_dir / "compatibility_snapshot_pg14"
+    compatibility_snapshot_dir = (
+        test_output_dir.parent / "test_create_snapshot" / "compatibility_snapshot_pg14"
+    )
 
     breaking_changes_allowed = (
         os.environ.get("ALLOW_FORWARD_COMPATIBILITY_BREAKAGE", "false").lower() == "true"

test_runner/regress/test_disk_usage_eviction.py

@@ -136,7 +136,9 @@ def eviction_env(request, neon_env_builder: NeonEnvBuilder, pg_bin: PgBin) -> Ev
     env.pageserver.allowed_errors.append(r".* running disk usage based eviction due to pressure.*")
 
     # remove the initial tenant
+    ## why wait for upload queue? => https://github.com/neondatabase/neon/issues/3865
     assert env.initial_timeline
+    wait_for_upload_queue_empty(pageserver_http, env.initial_tenant, env.initial_timeline)
     pageserver_http.tenant_detach(env.initial_tenant)
     assert isinstance(env.remote_storage, LocalFsStorage)
     tenant_remote_storage = env.remote_storage.root / "tenants" / str(env.initial_tenant)

test_runner/regress/test_duplicate_layers.py

@@ -1,42 +0,0 @@
-import time
-
-import pytest
-from fixtures.neon_fixtures import NeonEnvBuilder, PgBin
-
-
-# Test duplicate layer detection
-#
-# This test sets fail point at the end of first compaction phase:
-# after flushing new L1 layers but before deletion of L0 layes
-# It should cause generation of duplicate L1 layer by compaction after restart
-@pytest.mark.timeout(600)
-def test_duplicate_layers(neon_env_builder: NeonEnvBuilder, pg_bin: PgBin):
-    env = neon_env_builder.init_start()
-
-    # These warnings are expected, when the pageserver is restarted abruptly
-    env.pageserver.allowed_errors.append(".*found future image layer.*")
-    env.pageserver.allowed_errors.append(".*found future delta layer.*")
-    env.pageserver.allowed_errors.append(".*duplicate layer.*")
-
-    pageserver_http = env.pageserver.http_client()
-
-    # Use aggressive compaction and checkpoint settings
-    tenant_id, _ = env.neon_cli.create_tenant(
-        conf={
-            "checkpoint_distance": f"{1024 ** 2}",
-            "compaction_target_size": f"{1024 ** 2}",
-            "compaction_period": "1 s",
-            "compaction_threshold": "3",
-        }
-    )
-    endpoint = env.endpoints.create_start("main", tenant_id=tenant_id)
-    connstr = endpoint.connstr(options="-csynchronous_commit=off")
-    pg_bin.run_capture(["pgbench", "-i", "-s10", connstr])
-
-    pageserver_http.configure_failpoints(("compact-level0-phase1-finish", "exit"))
-
-    with pytest.raises(Exception):
-        pg_bin.run_capture(["pgbench", "-P1", "-N", "-c5", "-T500", "-Mprepared", connstr])
-    env.pageserver.stop()
-    env.pageserver.start()
-    time.sleep(10)  # let compaction to be performed

test_runner/regress/test_hot_standby.py

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

test_runner/regress/test_pg_regress.py

@@ -5,7 +5,6 @@ from pathlib import Path
 
 import pytest
 from fixtures.neon_fixtures import NeonEnv, check_restored_datadir_content
-from fixtures.pg_version import PgVersion, xfail_on_postgres
 
 
 # Run the main PostgreSQL regression tests, in src/test/regress.
@@ -72,7 +71,6 @@ def test_pg_regress(
 #
 # This runs for a long time, especially in debug mode, so use a larger-than-default
 # timeout.
-@xfail_on_postgres(PgVersion.V15, reason="https://github.com/neondatabase/neon/pull/4213")
 @pytest.mark.timeout(1800)
 def test_isolation(
     neon_simple_env: NeonEnv,

test_runner/regress/test_remote_storage.py

@@ -2,12 +2,11 @@
 # env NEON_PAGESERVER_OVERRIDES="remote_storage={local_path='/tmp/neon_zzz/'}" poetry ......
 
 import os
-import queue
 import shutil
 import threading
 import time
 from pathlib import Path
-from typing import Dict, List, Optional, Tuple
+from typing import Dict, List, Tuple
 
 import pytest
 from fixtures.log_helper import log
@@ -27,7 +26,6 @@ from fixtures.pageserver.utils import (
 )
 from fixtures.types import Lsn, TenantId, TimelineId
 from fixtures.utils import print_gc_result, query_scalar, wait_until
-from requests import ReadTimeout
 
 
 #
@@ -628,7 +626,10 @@ def test_empty_branch_remote_storage_upload(
 
     new_branch_name = "new_branch"
     new_branch_timeline_id = env.neon_cli.create_branch(new_branch_name, "main", env.initial_tenant)
-    assert_nothing_to_upload(client, env.initial_tenant, new_branch_timeline_id)
+
+    with env.endpoints.create_start(new_branch_name, tenant_id=env.initial_tenant) as endpoint:
+        wait_for_last_flush_lsn(env, endpoint, env.initial_tenant, new_branch_timeline_id)
+    wait_upload_queue_empty(client, env.initial_tenant, new_branch_timeline_id)
 
     timelines_before_detach = set(
         map(
@@ -657,19 +658,13 @@ def test_empty_branch_remote_storage_upload(
     ), f"Expected to have same timelines after reattach, but got {timelines_after_detach}"
 
 
+# Branches off a root branch, but does not write anything to the new branch, so it has a metadata file only.
+# Ensures the branch is not on the remote storage and restarts the pageserver — the branch should be uploaded after the restart.
 @pytest.mark.parametrize("remote_storage_kind", [RemoteStorageKind.LOCAL_FS])
 def test_empty_branch_remote_storage_upload_on_restart(
     neon_env_builder: NeonEnvBuilder,
     remote_storage_kind: RemoteStorageKind,
 ):
-    """
-    Branches off a root branch, but does not write anything to the new branch, so
-    it has a metadata file only.
-
-    Ensures the branch is not on the remote storage and restarts the pageserver
-    — the upload should be scheduled by load, and create_timeline should await
-    for it even though it gets 409 Conflict.
-    """
     neon_env_builder.enable_remote_storage(
         remote_storage_kind=remote_storage_kind,
         test_name="test_empty_branch_remote_storage_upload_on_restart",
@@ -678,87 +673,35 @@ def test_empty_branch_remote_storage_upload_on_restart(
     env = neon_env_builder.init_start()
     client = env.pageserver.http_client()
 
-    client.configure_failpoints(("before-upload-index", "return"))
+    new_branch_name = "new_branch"
+    new_branch_timeline_id = env.neon_cli.create_branch(new_branch_name, "main", env.initial_tenant)
 
-    new_branch_timeline_id = TimelineId.generate()
+    with env.endpoints.create_start(new_branch_name, tenant_id=env.initial_tenant) as endpoint:
+        wait_for_last_flush_lsn(env, endpoint, env.initial_tenant, new_branch_timeline_id)
+    wait_upload_queue_empty(client, env.initial_tenant, new_branch_timeline_id)
 
-    with pytest.raises(ReadTimeout):
-        client.timeline_create(
-            tenant_id=env.initial_tenant,
-            ancestor_timeline_id=env.initial_timeline,
-            new_timeline_id=new_branch_timeline_id,
-            pg_version=env.pg_version,
-            timeout=4,
-        )
-
-    env.pageserver.allowed_errors.append(
-        f".*POST.* path=/v1/tenant/{env.initial_tenant}/timeline.* request was dropped before completing"
-    )
-
-    # index upload is now hitting the failpoint, should not block the shutdown
     env.pageserver.stop()
 
-    timeline_path = (
-        Path("tenants") / str(env.initial_tenant) / "timelines" / str(new_branch_timeline_id)
-    )
-
-    local_metadata = env.repo_dir / timeline_path / "metadata"
-    assert local_metadata.is_file(), "timeout cancelled timeline branching, not the upload"
-
+    # Remove new branch from the remote storage
     assert isinstance(env.remote_storage, LocalFsStorage)
-    new_branch_on_remote_storage = env.remote_storage.root / timeline_path
+    new_branch_on_remote_storage = (
+        env.remote_storage.root
+        / "tenants"
+        / str(env.initial_tenant)
+        / "timelines"
+        / str(new_branch_timeline_id)
+    )
     assert (
-        not new_branch_on_remote_storage.exists()
-    ), "failpoint should had prohibited index_part.json upload"
+        new_branch_on_remote_storage.is_dir()
+    ), f"'{new_branch_on_remote_storage}' path does not exist on the remote storage"
+    shutil.rmtree(new_branch_on_remote_storage)
 
-    # during reconciliation we should had scheduled the uploads and on the
-    # retried create_timeline, we will await for those to complete on next
-    # client.timeline_create
-    env.pageserver.start(extra_env_vars={"FAILPOINTS": "before-upload-index=return"})
+    env.pageserver.start()
 
-    # sleep a bit to force the upload task go into exponential backoff
-    time.sleep(1)
-
-    q: queue.Queue[Optional[PageserverApiException]] = queue.Queue()
-    barrier = threading.Barrier(2)
-
-    def create_in_background():
-        barrier.wait()
-        try:
-            client.timeline_create(
-                tenant_id=env.initial_tenant,
-                ancestor_timeline_id=env.initial_timeline,
-                new_timeline_id=new_branch_timeline_id,
-                pg_version=env.pg_version,
-            )
-            q.put(None)
-        except PageserverApiException as e:
-            q.put(e)
-
-    create_thread = threading.Thread(target=create_in_background)
-    create_thread.start()
-
-    try:
-        # maximize chances of actually waiting for the uploads by create_timeline
-        barrier.wait()
-
-        assert not new_branch_on_remote_storage.exists(), "failpoint should had stopped uploading"
-
-        client.configure_failpoints(("before-upload-index", "off"))
-        conflict = q.get()
-
-        assert conflict, "create_timeline should not have succeeded"
-        assert (
-            conflict.status_code == 409
-        ), "timeline was created before restart, and uploads scheduled during initial load, so we expect 409 conflict"
-
-        assert_nothing_to_upload(client, env.initial_tenant, new_branch_timeline_id)
-
-        assert (
-            new_branch_on_remote_storage / "index_part.json"
-        ).is_file(), "uploads scheduled during initial load should had been awaited for"
-    finally:
-        create_thread.join()
+    wait_upload_queue_empty(client, env.initial_tenant, new_branch_timeline_id)
+    assert (
+        new_branch_on_remote_storage.is_dir()
+    ), f"New branch should have been reuploaded on pageserver restart to the remote storage path '{new_branch_on_remote_storage}'"
 
 
 def wait_upload_queue_empty(
@@ -809,17 +752,4 @@ def get_queued_count(
     return int(val)
 
 
-def assert_nothing_to_upload(
-    client: PageserverHttpClient,
-    tenant_id: TenantId,
-    timeline_id: TimelineId,
-):
-    """
-    Check last_record_lsn == remote_consistent_lsn. Assert works only for empty timelines, which
-    do not have anything to compact or gc.
-    """
-    detail = client.timeline_detail(tenant_id, timeline_id)
-    assert Lsn(detail["last_record_lsn"]) == Lsn(detail["remote_consistent_lsn"])
-
-
 # TODO Test that we correctly handle GC of files that are stuck in upload queue.

test_runner/regress/test_tenant_relocation.py

@@ -1,7 +1,5 @@
 import os
-import shutil
 import threading
-import time
 from contextlib import closing, contextmanager
 from pathlib import Path
 from typing import Any, Dict, Optional, Tuple
@@ -14,8 +12,6 @@ from fixtures.neon_fixtures import (
     NeonEnv,
     NeonEnvBuilder,
     PortDistributor,
-    RemoteStorageKind,
-    available_remote_storages,
 )
 from fixtures.pageserver.http import PageserverHttpClient
 from fixtures.pageserver.utils import (
@@ -516,225 +512,3 @@ def test_tenant_relocation(
             if line.startswith("listen_pg_addr"):
                 lines[i] = f"listen_pg_addr = 'localhost:{env.pageserver.service_port.pg}'"
         (env.repo_dir / "config").write_text("\n".join(lines))
-
-
-# Simulate hard crash of pageserver and re-attach a tenant with a branch
-#
-# This exercises a race condition after tenant attach, where the
-# branch point on the ancestor timeline is greater than the ancestor's
-# last-record LSN. We had a bug where GetPage incorrectly followed the
-# timeline to the ancestor without waiting for the missing WAL to
-# arrive.
-@pytest.mark.parametrize("remote_storage_kind", available_remote_storages())
-def test_emergency_relocate_with_branches_slow_replay(
-    neon_env_builder: NeonEnvBuilder,
-    remote_storage_kind: RemoteStorageKind,
-):
-    neon_env_builder.enable_remote_storage(
-        remote_storage_kind=remote_storage_kind,
-        test_name="test_emergency_relocate_with_branches_slow_replay",
-    )
-
-    env = neon_env_builder.init_start()
-    env.pageserver.is_testing_enabled_or_skip()
-    pageserver_http = env.pageserver.http_client()
-
-    # Prepare for the test:
-    #
-    # - Main branch, with a table and two inserts to it.
-    # - A logical replication message between the inserts, so that we can conveniently
-    #   pause the WAL ingestion between the two inserts.
-    # - Child branch, created after the inserts
-    tenant_id, _ = env.neon_cli.create_tenant()
-
-    main_endpoint = env.endpoints.create_start("main", tenant_id=tenant_id)
-    with main_endpoint.cursor() as cur:
-        cur.execute("CREATE TABLE test_reattach (t text)")
-        cur.execute("INSERT INTO test_reattach VALUES ('before pause')")
-
-        cur.execute("SELECT pg_logical_emit_message(false, 'neon-test', 'between inserts')")
-
-        cur.execute("INSERT INTO test_reattach VALUES ('after pause')")
-        current_lsn = Lsn(query_scalar(cur, "SELECT pg_current_wal_flush_lsn()"))
-
-    main_endpoint.stop()
-    env.neon_cli.create_branch("child", tenant_id=tenant_id, ancestor_start_lsn=current_lsn)
-
-    # Now kill the pageserver, remove the tenant directory, and restart. This simulates
-    # the scenario that a pageserver dies unexpectedly and cannot be recovered, so we relocate
-    # the tenant to a different pageserver. We reuse the same pageserver because it's
-    # simpler than initializing a new one from scratch, but the effect on the single tenant
-    # is the same.
-    env.pageserver.stop(immediate=True)
-    shutil.rmtree(Path(env.repo_dir) / "tenants" / str(tenant_id))
-    env.pageserver.start()
-
-    # This fail point will pause the WAL ingestion on the main branch, after the
-    # the first insert
-    pageserver_http.configure_failpoints([("wal-ingest-logical-message-sleep", "return(5000)")])
-
-    # Attach and wait a few seconds to give it time to load the tenants, attach to the
-    # safekeepers, and to stream and ingest the WAL up to the pause-point.
-    before_attach_time = time.time()
-    pageserver_http.tenant_attach(tenant_id)
-    time.sleep(3)
-
-    # The wal ingestion on the main timeline should now be paused at the fail point.
-    # Run a query on the child branch. The GetPage requests for this should recurse to the
-    # parent timeline, and wait for the WAL to be ingested there. Otherwise it won't see
-    # the second insert.
-    child_endpoint = env.endpoints.create_start("child", tenant_id=tenant_id)
-    with child_endpoint.cursor() as cur:
-        cur.execute("SELECT * FROM test_reattach")
-        assert cur.fetchall() == [("before pause",), ("after pause",)]
-
-    # Sanity check that the failpoint was reached
-    assert env.pageserver.log_contains('failpoint "wal-ingest-logical-message-sleep": sleep done')
-    assert time.time() - before_attach_time > 5
-
-    # Clean up
-    pageserver_http.configure_failpoints(("wal-ingest-logical-message-sleep", "off"))
-
-
-# Simulate hard crash of pageserver and re-attach a tenant with a branch
-#
-# This exercises the same race condition after as
-# 'test_emergency_relocate_with_branches_slow_replay', but this test case
-# is closer to the original scenario where we originally found the
-# issue.
-#
-# In this scenario, the incorrect result to get-request leads to
-# *permanent damage* in the child timeline, because ingesting the WAL
-# on the child timeline depended on incorrect view of the parent. This
-# test reproduced one such case; the symptom was an error on the child, when
-# trying to connect to the child endpoint after re-attaching the tenant:
-#
-# FATAL: database "neondb" does not exist
-#
-# In the original case where we bumped into this, the error was slightly
-# different:
-#
-# FATAL:  "base/16385" is not a valid data directory
-# DETAIL:  File "base/16385/PG_VERSION" is missing.
-#
-### Detailed explanation of the original bug and why it lead to that error:
-#
-# The WAL on the main and the child branches look like this:
-#
-#    Main                                  Child
-# 1. CREATE DATABASE
-#           <child branch is created>
-# 2. CREATE TABLE AS SELECT ...            3. CREATE TABLE AS SELECT ...
-#
-# None of these WAL records have been flushed to disk or uploaded to remote
-# storage in the pageserver yet, when the tenant is detached.
-#
-# After detach and re-attach, a walreceiver is spawned on both timelines.
-# They will connect to the safekeepers and start ingesting the WAL
-# from their respective IndexParts' `disk_consistent_lsn` onward.
-#
-# The bug occurs if the child branch's walreceiver runs before the
-# main's.  It receives the SMGR_CREATE WAL record emitted by the
-# CREATE TABLE statement (3.), and applies it, without seeing the
-# effect of the CREATE DATABASE statement.
-#
-# To understand why that leads to a 'File "base/16385/PG_VERSION" is
-# missing' error, let's look at what the handlers for the WAL records
-# do:
-#
-# CREATE DATABASE WAL record is handled by ingest_xlog_dbase_create:
-#
-#    ingest_xlog_dbase_create:
-#     put_relmap_file()
-#       // NOTE 'true': It means that there is a relmapper and PG_VERSION file
-# 1:    let r = dbdir.dbdirs.insert((spcnode, dbnode), true);
-#
-#
-# CREATE TABLE emits an SMGR_CREATE record, which is handled by:
-#
-#    ingest_xlog_smgr_create:
-#      put_rel_creation:
-#        ...
-#        let mut rel_dir = if dbdir.dbdirs.get(&(rel.spcnode, rel.dbnode)).is_none() {
-# 2:         // Didn't exist. Update dbdir
-#            dbdir.dbdirs.insert((rel.spcnode, rel.dbnode), false);
-#            let buf = DbDirectory::ser(&dbdir)?;
-#            self.put(DBDIR_KEY, Value::Image(buf.into()));
-#
-#            // and create the RelDirectory
-#            RelDirectory::default()
-#        } else {
-# 3:         // reldir already exists, fetch it
-#            RelDirectory::des(&self.get(rel_dir_key, ctx).await?)?
-#        };
-#
-#
-# In the correct ordering, the SMGR_CREATE record is applied after the
-# CREATE DATABASE record. The CREATE DATABASE creates the entry in the
-# 'dbdir', with the 'true' flag that indicates that PG_VERSION exists
-# (1). The SMGR_CREATE handler calls put_rel_creation, which finds the
-# dbdir entry that the CREATE DATABASE record created, and takes the
-# "reldir already exists, fetch it" else-branch at the if statement (3).
-#
-# In the incorrect ordering, the child walreceiver applies the
-# SMGR_CREATE record without seeing the effects of the CREATE
-# DATABASE. In that case, put_rel_creation takes the "Didn't
-# exist. Update dbir" path (2), and inserts an entry in the
-# DbDirectory with 'false' to indicate there is no PG_VERSION file.
-#
-@pytest.mark.parametrize("remote_storage_kind", available_remote_storages())
-def test_emergency_relocate_with_branches_createdb(
-    neon_env_builder: NeonEnvBuilder,
-    remote_storage_kind: RemoteStorageKind,
-):
-    neon_env_builder.enable_remote_storage(
-        remote_storage_kind=remote_storage_kind,
-        test_name="test_emergency_relocate_with_branches_createdb",
-    )
-
-    env = neon_env_builder.init_start()
-    pageserver_http = env.pageserver.http_client()
-
-    # create new nenant
-    tenant_id, _ = env.neon_cli.create_tenant()
-
-    main_endpoint = env.endpoints.create_start("main", tenant_id=tenant_id)
-    with main_endpoint.cursor() as cur:
-        cur.execute("SELECT pg_logical_emit_message(false, 'neon-test', 'between inserts')")
-
-        cur.execute("CREATE DATABASE neondb")
-        current_lsn = Lsn(query_scalar(cur, "SELECT pg_current_wal_flush_lsn()"))
-    env.neon_cli.create_branch("child", tenant_id=tenant_id, ancestor_start_lsn=current_lsn)
-
-    with main_endpoint.cursor(dbname="neondb") as cur:
-        cur.execute("CREATE TABLE test_migrate_one AS SELECT generate_series(1,100)")
-    main_endpoint.stop()
-
-    child_endpoint = env.endpoints.create_start("child", tenant_id=tenant_id)
-    with child_endpoint.cursor(dbname="neondb") as cur:
-        cur.execute("CREATE TABLE test_migrate_one AS SELECT generate_series(1,200)")
-    child_endpoint.stop()
-
-    # Kill the pageserver, remove the tenant directory, and restart
-    env.pageserver.stop(immediate=True)
-    shutil.rmtree(Path(env.repo_dir) / "tenants" / str(tenant_id))
-    env.pageserver.start()
-
-    # Wait before ingesting the WAL for CREATE DATABASE on the main branch. The original
-    # bug reproduced easily even without this, as there is always some delay between
-    # loading the timeline and establishing the connection to the safekeeper to stream and
-    # ingest the WAL, but let's make this less dependent on accidental timing.
-    pageserver_http.configure_failpoints([("wal-ingest-logical-message-sleep", "return(5000)")])
-    before_attach_time = time.time()
-    pageserver_http.tenant_attach(tenant_id)
-
-    child_endpoint.start()
-    with child_endpoint.cursor(dbname="neondb") as cur:
-        assert query_scalar(cur, "SELECT count(*) FROM test_migrate_one") == 200
-
-    # Sanity check that the failpoint was reached
-    assert env.pageserver.log_contains('failpoint "wal-ingest-logical-message-sleep": sleep done')
-    assert time.time() - before_attach_time > 5
-
-    # Clean up
-    pageserver_http.configure_failpoints(("wal-ingest-logical-message-sleep", "off"))

test_runner/regress/test_tenant_size.py

@@ -11,7 +11,6 @@ from fixtures.neon_fixtures import (
     wait_for_wal_insert_lsn,
 )
 from fixtures.pageserver.http import PageserverHttpClient
-from fixtures.pg_version import PgVersion, xfail_on_postgres
 from fixtures.types import Lsn, TenantId, TimelineId
 
 
@@ -513,7 +512,6 @@ def test_single_branch_get_tenant_size_grows(
     assert size_after == prev, "size after restarting pageserver should not have changed"
 
 
-@xfail_on_postgres(PgVersion.V15, reason="Test significantly more flaky on Postgres 15")
 def test_get_tenant_size_with_multiple_branches(
     neon_env_builder: NeonEnvBuilder, test_output_dir: Path
 ):

test_runner/regress/test_timeline_delete.py

@@ -346,20 +346,31 @@ def test_concurrent_timeline_delete_if_first_stuck_at_index_upload(
     failpoint_name = "persist_index_part_with_deleted_flag_after_set_before_upload_pause"
     ps_http.configure_failpoints((failpoint_name, "pause"))
 
-    def first_call(result_queue):
+    def delete_timeline_call(name, result_queue, barrier):
+        if barrier:
+            barrier.wait()
         try:
-            log.info("first call start")
+            log.info(f"{name} call start")
             ps_http.timeline_delete(env.initial_tenant, child_timeline_id, timeout=10)
-            log.info("first call success")
+            log.info(f"{name} call success")
             result_queue.put("success")
         except Exception:
-            log.exception("first call failed")
+            log.exception(f"{name} call failed")
             result_queue.put("failure, see log for stack trace")
 
-    first_call_result: queue.Queue[str] = queue.Queue()
-    first_call_thread = threading.Thread(target=first_call, args=(first_call_result,))
+    delete_results: queue.Queue[str] = queue.Queue()
+    first_call_thread = threading.Thread(
+        target=delete_timeline_call,
+        args=(
+            "1st",
+            delete_results,
+            None,
+        ),
+    )
     first_call_thread.start()
 
+    second_call_thread = None
+
     try:
 
         def first_call_hit_failpoint():
@@ -369,30 +380,42 @@ def test_concurrent_timeline_delete_if_first_stuck_at_index_upload(
 
         wait_until(50, 0.1, first_call_hit_failpoint)
 
-        # make the second call and assert behavior
-        log.info("second call start")
-        error_msg_re = "another task is already setting the deleted_flag, started at"
-        with pytest.raises(PageserverApiException, match=error_msg_re) as second_call_err:
-            ps_http.timeline_delete(env.initial_tenant, child_timeline_id)
-        assert second_call_err.value.status_code == 500
-        env.pageserver.allowed_errors.append(f".*{child_timeline_id}.*{error_msg_re}.*")
-        # the second call will try to transition the timeline into Stopping state as well
+        barrier = threading.Barrier(2)
+        second_call_thread = threading.Thread(
+            target=delete_timeline_call,
+            args=(
+                "2nd",
+                delete_results,
+                barrier,
+            ),
+        )
+        second_call_thread.start()
+
+        barrier.wait()
+
+        # release the pause
+        ps_http.configure_failpoints((failpoint_name, "off"))
+
+        # both should had succeeded: the second call will coalesce with the already-ongoing first call
+        result = delete_results.get()
+        assert result == "success"
+        result = delete_results.get()
+        assert result == "success"
+
+        # the second call will try to transition the timeline into Stopping state, but it's already in that state
+        # (the transition to Stopping state is not part of the request coalescing, because Tenant and Timeline states are a mess already)
         env.pageserver.allowed_errors.append(
             f".*{child_timeline_id}.*Ignoring new state, equal to the existing one: Stopping"
         )
-        log.info("second call failed as expected")
-
-        # by now we know that the second call failed, let's ensure the first call will finish
-        ps_http.configure_failpoints((failpoint_name, "off"))
-
-        result = first_call_result.get()
-        assert result == "success"
-
     finally:
-        log.info("joining first call thread")
+        log.info("joining 1st thread")
         # in any case, make sure the lifetime of the thread is bounded to this test
         first_call_thread.join()
 
+        if second_call_thread:
+            log.info("joining 2nd thread")
+            second_call_thread.join()
+
 
 def test_delete_timeline_client_hangup(neon_env_builder: NeonEnvBuilder):
     """
@@ -440,9 +463,16 @@ def test_delete_timeline_client_hangup(neon_env_builder: NeonEnvBuilder):
     # ok, retry without failpoint, it should succeed
     ps_http.configure_failpoints((failpoint_name, "off"))
 
-    # this should succeed
-    ps_http.timeline_delete(env.initial_tenant, child_timeline_id, timeout=2)
-    # the second call will try to transition the timeline into Stopping state, but it's already in that state
-    env.pageserver.allowed_errors.append(
-        f".*{child_timeline_id}.*Ignoring new state, equal to the existing one: Stopping"
-    )
+    try:
+        ps_http.timeline_delete(env.initial_tenant, child_timeline_id)
+        env.pageserver.allowed_errors.append(
+            f".*{child_timeline_id}.*Ignoring new state, equal to the existing one: Stopping"
+        )
+    except PageserverApiException as e:
+        if e.status_code != 404:
+            raise e
+        else:
+            # mock_s3 was fast enough to delete before we got the request in
+            env.pageserver.allowed_errors.append(
+                f".*{child_timeline_id}.*Error processing HTTP request: NotFound: timeline not found"
+            )

test_runner/regress/test_wal_restore.py

@@ -1,20 +1,14 @@
-import sys
 from pathlib import Path
 
-import pytest
 from fixtures.neon_fixtures import (
     NeonEnvBuilder,
     PgBin,
     PortDistributor,
     VanillaPostgres,
 )
-from fixtures.types import TenantId, TimelineId
+from fixtures.types import TenantId
 
 
-@pytest.mark.skipif(
-    sys.platform != "linux",
-    reason="restore_from_wal.sh supports only Linux",
-)
 def test_wal_restore(
     neon_env_builder: NeonEnvBuilder,
     pg_bin: PgBin,
@@ -28,7 +22,6 @@ def test_wal_restore(
     endpoint = env.endpoints.create_start("test_wal_restore")
     endpoint.safe_psql("create table t as select generate_series(1,300000)")
     tenant_id = TenantId(endpoint.safe_psql("show neon.tenant_id")[0][0])
-    timeline_id = TimelineId(endpoint.safe_psql("show neon.timeline_id")[0][0])
     env.neon_cli.pageserver_stop()
     port = port_distributor.get_port()
     data_dir = test_output_dir / "pgsql.restored"
@@ -37,16 +30,9 @@ def test_wal_restore(
     ) as restored:
         pg_bin.run_capture(
             [
-                str(base_dir / "libs" / "utils" / "scripts" / "restore_from_wal.sh"),
+                str(base_dir / "libs/utils/scripts/restore_from_wal.sh"),
                 str(pg_distrib_dir / f"v{env.pg_version}/bin"),
-                str(
-                    test_output_dir
-                    / "repo"
-                    / "safekeepers"
-                    / "sk1"
-                    / str(tenant_id)
-                    / str(timeline_id)
-                ),
+                str(test_output_dir / "repo" / "safekeepers" / "sk1" / str(tenant_id) / "*"),
                 str(data_dir),
                 str(port),
             ]