Do vacuum freeze before copying data

.cargo/config.toml

@@ -12,11 +12,6 @@ opt-level = 3
 # Turn on a small amount of optimization in Development mode.
 opt-level = 1
 
-[build]
-# This is only present for local builds, as it will be overridden
-# by the RUSTDOCFLAGS env var in CI.
-rustdocflags = ["-Arustdoc::private_intra_doc_links"]
-
 [alias]
 build_testing = ["build", "--features", "testing"]
 neon = ["run", "--bin", "neon_local"]

.dockerignore

@@ -18,7 +18,6 @@
 !trace/
 !vendor/postgres-v14/
 !vendor/postgres-v15/
-!vendor/postgres-v16/
 !workspace_hack/
 !neon_local/
 !scripts/ninstall.sh

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

@@ -105,7 +105,7 @@ runs:
         # 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')
+        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

.github/workflows/approved-for-ci-run.yml

@@ -1,55 +0,0 @@
-name: Handle `approved-for-ci-run` label
-# This workflow helps to run CI pipeline for PRs made by external contributors (from forks).
-
-on:
-  pull_request:
-    types:
-      # Default types that triggers a workflow ([1]):
-      # - [1] https://docs.github.com/en/actions/using-workflows/events-that-trigger-workflows#pull_request
-      - opened
-      - synchronize
-      - reopened
-      # Types that we wand to handle in addition to keep labels tidy:
-      - closed
-      # Actual magic happens here:
-      - labeled
-
-env:
-  GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-  PR_NUMBER: ${{ github.event.pull_request.number }}
-
-jobs:
-  remove-label:
-    # Remove `approved-for-ci-run` label if the workflow is triggered by changes in a PR.
-    # The PR should be reviewed and labelled manually again.
-
-    runs-on: [ ubuntu-latest ]
-
-    if: |
-      contains(fromJSON('["opened", "synchronize", "reopened", "closed"]'), github.event.action) &&
-      contains(github.event.pull_request.labels.*.name, 'approved-for-ci-run')
-
-    steps:
-      - run: gh pr --repo "${GITHUB_REPOSITORY}" edit "${PR_NUMBER}" --remove-label "approved-for-ci-run"
-
-  create-branch:
-    # Create a local branch for an `approved-for-ci-run` labelled PR to run CI pipeline in it.
-
-    runs-on: [ ubuntu-latest ]
-
-    if: |
-      github.event.action == 'labeled' &&
-      contains(github.event.pull_request.labels.*.name, 'approved-for-ci-run')
-
-    steps:
-      - run: gh pr --repo "${GITHUB_REPOSITORY}" edit "${PR_NUMBER}" --remove-label "approved-for-ci-run"
-
-      - uses: actions/checkout@v3
-        with:
-          ref: main
-
-      - run: gh pr checkout "${PR_NUMBER}"
-
-      - run: git checkout -b "ci-run/pr-${PR_NUMBER}"
-
-      - run: git push --force origin "ci-run/pr-${PR_NUMBER}"

.github/workflows/build_and_test.yml

@@ -5,7 +5,6 @@ on:
     branches:
       - main
       - release
-      - ci-run/pr-*
   pull_request:
 
 defaults:
@@ -128,11 +127,6 @@ jobs:
       - name: Run cargo clippy (release)
         run: cargo hack --feature-powerset clippy --release $CLIPPY_COMMON_ARGS
 
-      - name: Check documentation generation
-        run: cargo doc --workspace --no-deps --document-private-items
-        env:
-            RUSTDOCFLAGS: "-Dwarnings -Arustdoc::private_intra_doc_links"
-
       # Use `${{ !cancelled() }}` to run quck tests after the longer clippy run
       - name: Check formatting
         if: ${{ !cancelled() }}
@@ -161,7 +155,7 @@ jobs:
         build_type: [ debug, release ]
     env:
       BUILD_TYPE: ${{ matrix.build_type }}
-      GIT_VERSION: ${{ github.event.pull_request.head.sha || github.sha }}
+      GIT_VERSION: ${{ github.sha }}
 
     steps:
       - name: Fix git ownership
@@ -180,27 +174,6 @@ jobs:
           submodules: true
           fetch-depth: 1
 
-      - name: Check Postgres submodules revision
-        shell: bash -euo pipefail {0}
-        run: |
-          # This is a temporary solution to ensure that the Postgres submodules revision is correct (i.e. the updated intentionally).
-          # Eventually it will be replaced by a regression test https://github.com/neondatabase/neon/pull/4603
-
-          FAILED=false
-          for postgres in postgres-v14 postgres-v15; do
-            expected=$(cat vendor/revisions.json | jq --raw-output '."'"${postgres}"'"')
-            actual=$(git rev-parse "HEAD:vendor/${postgres}")
-            if [ "${expected}" != "${actual}" ]; then
-              echo >&2 "Expected ${postgres} rev to be at '${expected}', but it is at '${actual}'"
-              FAILED=true
-            fi
-          done
-
-          if [ "${FAILED}" = "true" ]; then
-            echo >&2 "Please update vendors/revisions.json if these changes are intentional"
-            exit 1
-          fi
-
       - name: Set pg 14 revision for caching
         id: pg_v14_rev
         run: echo pg_rev=$(git rev-parse HEAD:vendor/postgres-v14) >> $GITHUB_OUTPUT
@@ -641,7 +614,7 @@ jobs:
           /kaniko/executor --reproducible --snapshot-mode=redo --skip-unused-stages --cache=true
                            --cache-repo 369495373322.dkr.ecr.eu-central-1.amazonaws.com/cache
                            --context .
-                           --build-arg GIT_VERSION=${{ github.event.pull_request.head.sha || github.sha }}
+                           --build-arg GIT_VERSION=${{ github.sha }}
                            --build-arg REPOSITORY=369495373322.dkr.ecr.eu-central-1.amazonaws.com
                            --destination 369495373322.dkr.ecr.eu-central-1.amazonaws.com/neon:${{needs.tag.outputs.build-tag}}
                            --destination neondatabase/neon:${{needs.tag.outputs.build-tag}}
@@ -685,7 +658,7 @@ jobs:
           /kaniko/executor --reproducible --snapshot-mode=redo --skip-unused-stages --cache=true
                            --cache-repo 369495373322.dkr.ecr.eu-central-1.amazonaws.com/cache
                            --context .
-                           --build-arg GIT_VERSION=${{ github.event.pull_request.head.sha || github.sha }}
+                           --build-arg GIT_VERSION=${{ github.sha }}
                            --build-arg BUILD_TAG=${{needs.tag.outputs.build-tag}}
                            --build-arg REPOSITORY=369495373322.dkr.ecr.eu-central-1.amazonaws.com
                            --dockerfile Dockerfile.compute-tools
@@ -742,7 +715,7 @@ jobs:
           /kaniko/executor --reproducible --snapshot-mode=redo --skip-unused-stages --cache=true
                            --cache-repo 369495373322.dkr.ecr.eu-central-1.amazonaws.com/cache
                            --context .
-                           --build-arg GIT_VERSION=${{ github.event.pull_request.head.sha || github.sha }}
+                           --build-arg GIT_VERSION=${{ github.sha }}
                            --build-arg PG_VERSION=${{ matrix.version }}
                            --build-arg BUILD_TAG=${{needs.tag.outputs.build-tag}}
                            --build-arg REPOSITORY=369495373322.dkr.ecr.eu-central-1.amazonaws.com
@@ -769,7 +742,7 @@ jobs:
           /kaniko/executor --reproducible --snapshot-mode=redo --skip-unused-stages --cache=true \
                            --cache-repo 369495373322.dkr.ecr.eu-central-1.amazonaws.com/cache \
                            --context . \
-                           --build-arg GIT_VERSION=${{ github.event.pull_request.head.sha || github.sha }} \
+                           --build-arg GIT_VERSION=${{ github.sha }} \
                            --build-arg PG_VERSION=${{ matrix.version }} \
                            --build-arg BUILD_TAG=${{needs.tag.outputs.build-tag}} \
                            --build-arg REPOSITORY=369495373322.dkr.ecr.eu-central-1.amazonaws.com \
@@ -794,7 +767,7 @@ jobs:
       run:
         shell: sh -eu {0}
     env:
-      VM_BUILDER_VERSION: v0.13.1
+      VM_BUILDER_VERSION: v0.11.1
 
     steps:
       - name: Checkout

.github/workflows/neon_extra_builds.yml

@@ -3,8 +3,7 @@ name: Check neon with extra platform builds
 on:
   push:
     branches:
-      - main
-      - ci-run/pr-*
+    - main
   pull_request:
 
 defaults:

.gitmodules

@@ -6,7 +6,3 @@
 	path = vendor/postgres-v15
 	url = https://github.com/neondatabase/postgres.git
 	branch = REL_15_STABLE_neon
-[submodule "vendor/postgres-v16"]
-	path = vendor/postgres-v16
-	url = https://github.com/neondatabase/postgres.git
-	branch = REL_16_STABLE_neon

Cargo.lock

@@ -158,19 +158,6 @@ dependencies = [
  "syn 1.0.109",
 ]
 
-[[package]]
-name = "async-compression"
-version = "0.4.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5b0122885821398cc923ece939e24d1056a2384ee719432397fa9db87230ff11"
-dependencies = [
- "flate2",
- "futures-core",
- "memchr",
- "pin-project-lite",
- "tokio",
-]
-
 [[package]]
 name = "async-stream"
 version = "0.3.5"
@@ -606,7 +593,7 @@ dependencies = [
  "cc",
  "cfg-if",
  "libc",
- "miniz_oxide 0.6.2",
+ "miniz_oxide",
  "object",
  "rustc-demangle",
 ]
@@ -895,11 +882,9 @@ name = "compute_tools"
 version = "0.1.0"
 dependencies = [
  "anyhow",
- "async-compression",
  "chrono",
  "clap",
  "compute_api",
- "flate2",
  "futures",
  "hyper",
  "notify",
@@ -1382,16 +1367,6 @@ version = "0.4.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0ce7134b9999ecaf8bcd65542e436736ef32ddca1b3e06094cb6ec5755203b80"
 
-[[package]]
-name = "flate2"
-version = "1.0.26"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3b9429470923de8e8cbd4d2dc513535400b4b3fef0319fb5c4e1f520a7bef743"
-dependencies = [
- "crc32fast",
- "miniz_oxide 0.7.1",
-]
-
 [[package]]
 name = "fnv"
 version = "1.0.7"
@@ -2176,15 +2151,6 @@ dependencies = [
  "adler",
 ]
 
-[[package]]
-name = "miniz_oxide"
-version = "0.7.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e7810e0be55b428ada41041c41f32c9f1a42817901b4ccf45fa3d4b6561e74c7"
-dependencies = [
- "adler",
-]
-
 [[package]]
 name = "mio"
 version = "0.8.6"
@@ -2379,9 +2345,9 @@ dependencies = [
 
 [[package]]
 name = "opentelemetry"
-version = "0.19.0"
+version = "0.18.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5f4b8347cc26099d3aeee044065ecc3ae11469796b4d65d065a23a584ed92a6f"
+checksum = "69d6c3d7288a106c0a363e4b0e8d308058d56902adefb16f4936f417ffef086e"
 dependencies = [
  "opentelemetry_api",
  "opentelemetry_sdk",
@@ -2389,9 +2355,9 @@ dependencies = [
 
 [[package]]
 name = "opentelemetry-http"
-version = "0.8.0"
+version = "0.7.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a819b71d6530c4297b49b3cae2939ab3a8cc1b9f382826a1bc29dd0ca3864906"
+checksum = "1edc79add46364183ece1a4542592ca593e6421c60807232f5b8f7a31703825d"
 dependencies = [
  "async-trait",
  "bytes",
@@ -2402,9 +2368,9 @@ dependencies = [
 
 [[package]]
 name = "opentelemetry-otlp"
-version = "0.12.0"
+version = "0.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8af72d59a4484654ea8eb183fea5ae4eb6a41d7ac3e3bae5f4d2a282a3a7d3ca"
+checksum = "d1c928609d087790fc936a1067bdc310ae702bdf3b090c3f281b713622c8bbde"
 dependencies = [
  "async-trait",
  "futures",
@@ -2420,47 +2386,48 @@ dependencies = [
 
 [[package]]
 name = "opentelemetry-proto"
-version = "0.2.0"
+version = "0.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "045f8eea8c0fa19f7d48e7bc3128a39c2e5c533d5c61298c548dfefc1064474c"
+checksum = "d61a2f56df5574508dd86aaca016c917489e589ece4141df1b5e349af8d66c28"
 dependencies = [
  "futures",
  "futures-util",
  "opentelemetry",
  "prost",
  "tonic 0.8.3",
+ "tonic-build 0.8.4",
 ]
 
 [[package]]
 name = "opentelemetry-semantic-conventions"
-version = "0.11.0"
+version = "0.10.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "24e33428e6bf08c6f7fcea4ddb8e358fab0fe48ab877a87c70c6ebe20f673ce5"
+checksum = "9b02e0230abb0ab6636d18e2ba8fa02903ea63772281340ccac18e0af3ec9eeb"
 dependencies = [
  "opentelemetry",
 ]
 
 [[package]]
 name = "opentelemetry_api"
-version = "0.19.0"
+version = "0.18.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ed41783a5bf567688eb38372f2b7a8530f5a607a4b49d38dd7573236c23ca7e2"
+checksum = "c24f96e21e7acc813c7a8394ee94978929db2bcc46cf6b5014fc612bf7760c22"
 dependencies = [
  "fnv",
  "futures-channel",
  "futures-util",
  "indexmap",
+ "js-sys",
  "once_cell",
  "pin-project-lite",
  "thiserror",
- "urlencoding",
 ]
 
 [[package]]
 name = "opentelemetry_sdk"
-version = "0.19.0"
+version = "0.18.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8b3a2a91fdbfdd4d212c0dcc2ab540de2c2bcbbd90be17de7a7daf8822d010c1"
+checksum = "1ca41c4933371b61c2a2f214bf16931499af4ec90543604ec828f7a625c09113"
 dependencies = [
  "async-trait",
  "crossbeam-channel",
@@ -2515,7 +2482,6 @@ name = "pageserver"
 version = "0.1.0"
 dependencies = [
  "anyhow",
- "async-compression",
  "async-stream",
  "async-trait",
  "byteorder",
@@ -2532,7 +2498,6 @@ dependencies = [
  "enum-map",
  "enumset",
  "fail",
- "flate2",
  "futures",
  "git-version",
  "hex",
@@ -2936,9 +2901,9 @@ checksum = "dc375e1527247fe1a97d8b7156678dfe7c1af2fc075c9a4db3690ecd2a148068"
 
 [[package]]
 name = "proc-macro2"
-version = "1.0.64"
+version = "1.0.58"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "78803b62cbf1f46fde80d7c0e803111524b9877184cfe7c3033659490ac7a7da"
+checksum = "fa1fb82fc0c281dd9671101b66b771ebbe1eaf967b96ac8740dcba4b70005ca8"
 dependencies = [
  "unicode-ident",
 ]
@@ -3327,9 +3292,9 @@ dependencies = [
 
 [[package]]
 name = "reqwest-tracing"
-version = "0.4.5"
+version = "0.4.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1b97ad83c2fc18113346b7158d79732242002427c30f620fa817c1f32901e0a8"
+checksum = "783e8130d2427ddd7897dd3f814d4a3aea31b05deb42a4fdf8c18258fe5aefd1"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -3997,7 +3962,7 @@ dependencies = [
  "tokio",
  "tokio-stream",
  "tonic 0.9.2",
- "tonic-build",
+ "tonic-build 0.9.2",
  "tracing",
  "utils",
  "workspace_hack",
@@ -4515,6 +4480,19 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "tonic-build"
+version = "0.8.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5bf5e9b9c0f7e0a7c027dcfaba7b2c60816c7049171f679d99ee2ff65d0de8c4"
+dependencies = [
+ "prettyplease 0.1.25",
+ "proc-macro2",
+ "prost-build",
+ "quote",
+ "syn 1.0.109",
+]
+
 [[package]]
 name = "tonic-build"
 version = "0.9.2"
@@ -4638,9 +4616,9 @@ dependencies = [
 
 [[package]]
 name = "tracing-opentelemetry"
-version = "0.19.0"
+version = "0.18.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "00a39dcf9bfc1742fa4d6215253b33a6e474be78275884c216fc2a06267b3600"
+checksum = "21ebb87a95ea13271332df069020513ab70bdb5637ca42d6e492dc3bbbad48de"
 dependencies = [
  "once_cell",
  "opentelemetry",
@@ -4839,7 +4817,6 @@ dependencies = [
  "byteorder",
  "bytes",
  "chrono",
- "const_format",
  "criterion",
  "futures",
  "heapless",

Cargo.toml

@@ -32,8 +32,6 @@ license = "Apache-2.0"
 ## All dependency versions, used in the project
 [workspace.dependencies]
 anyhow = { version = "1.0", features = ["backtrace"] }
-async-compression = { version = "0.4.0", features = ["tokio", "gzip"] }
-flate2 = "1.0.26"
 async-stream = "0.3"
 async-trait = "0.1"
 aws-config = { version = "0.55", default-features = false, features=["rustls"] }
@@ -84,9 +82,9 @@ notify = "5.0.0"
 num_cpus = "1.15"
 num-traits = "0.2.15"
 once_cell = "1.13"
-opentelemetry = "0.19.0"
-opentelemetry-otlp = { version = "0.12.0", default_features=false, features = ["http-proto", "trace", "http", "reqwest-client"] }
-opentelemetry-semantic-conventions = "0.11.0"
+opentelemetry = "0.18.0"
+opentelemetry-otlp = { version = "0.11.0", default_features=false, features = ["http-proto", "trace", "http", "reqwest-client"] }
+opentelemetry-semantic-conventions = "0.10.0"
 parking_lot = "0.12"
 pbkdf2 = "0.12.1"
 pin-project-lite = "0.2"
@@ -95,7 +93,7 @@ prost = "0.11"
 rand = "0.8"
 regex = "1.4"
 reqwest = { version = "0.11", default-features = false, features = ["rustls-tls"] }
-reqwest-tracing = { version = "0.4.0", features = ["opentelemetry_0_19"] }
+reqwest-tracing = { version = "0.4.0", features = ["opentelemetry_0_18"] }
 reqwest-middleware = "0.2.0"
 reqwest-retry = "0.2.2"
 routerify = "3"
@@ -130,7 +128,7 @@ toml_edit = "0.19"
 tonic = {version = "0.9", features = ["tls", "tls-roots"]}
 tracing = "0.1"
 tracing-error = "0.2.0"
-tracing-opentelemetry = "0.19.0"
+tracing-opentelemetry = "0.18.0"
 tracing-subscriber = { version = "0.3", default_features = false, features = ["smallvec", "fmt", "tracing-log", "std", "env-filter"] }
 url = "2.2"
 uuid = { version = "1.2", features = ["v4", "serde"] }

Dockerfile

@@ -12,7 +12,6 @@ WORKDIR /home/nonroot
 
 COPY --chown=nonroot vendor/postgres-v14 vendor/postgres-v14
 COPY --chown=nonroot vendor/postgres-v15 vendor/postgres-v15
-COPY --chown=nonroot vendor/postgres-v16 vendor/postgres-v16
 COPY --chown=nonroot pgxn pgxn
 COPY --chown=nonroot Makefile Makefile
 COPY --chown=nonroot scripts/ninstall.sh scripts/ninstall.sh
@@ -40,7 +39,6 @@ ARG CACHEPOT_BUCKET=neon-github-dev
 
 COPY --from=pg-build /home/nonroot/pg_install/v14/include/postgresql/server pg_install/v14/include/postgresql/server
 COPY --from=pg-build /home/nonroot/pg_install/v15/include/postgresql/server pg_install/v15/include/postgresql/server
-COPY --from=pg-build /home/nonroot/pg_install/v16/include/postgresql/server pg_install/v16/include/postgresql/server
 COPY --chown=nonroot . .
 
 # Show build caching stats to check if it was used in the end.
@@ -81,7 +79,6 @@ COPY --from=build --chown=neon:neon /home/nonroot/target/release/proxy
 
 COPY --from=pg-build /home/nonroot/pg_install/v14 /usr/local/v14/
 COPY --from=pg-build /home/nonroot/pg_install/v15 /usr/local/v15/
-COPY --from=pg-build /home/nonroot/pg_install/v16 /usr/local/v16/
 COPY --from=pg-build /home/nonroot/postgres_install.tar.gz /data/
 
 # By default, pageserver uses `.neon/` working directory in WORKDIR, so create one and fill it with the dummy config.

Dockerfile.compute-node

@@ -132,20 +132,10 @@ RUN wget https://github.com/plv8/plv8/archive/refs/tags/v3.1.5.tar.gz -O plv8.ta
 FROM build-deps AS h3-pg-build
 COPY --from=pg-build /usr/local/pgsql/ /usr/local/pgsql/
 
-RUN case "$(uname -m)" in \
-      "x86_64") \
-        export CMAKE_CHECKSUM=739d372726cb23129d57a539ce1432453448816e345e1545f6127296926b6754 \
-        ;; \
-      "aarch64") \
-        export CMAKE_CHECKSUM=281b42627c9a1beed03e29706574d04c6c53fae4994472e90985ef018dd29c02 \
-        ;; \
-      *) \
-        echo "Unsupported architecture '$(uname -m)'. Supported are x86_64 and aarch64" && exit 1 \
-        ;; \
-    esac && \
-    wget https://github.com/Kitware/CMake/releases/download/v3.24.2/cmake-3.24.2-linux-$(uname -m).sh \
+# packaged cmake is too old
+RUN wget https://github.com/Kitware/CMake/releases/download/v3.24.2/cmake-3.24.2-linux-x86_64.sh \
       -q -O /tmp/cmake-install.sh \
-      && echo "${CMAKE_CHECKSUM} /tmp/cmake-install.sh" | sha256sum --check \
+      && echo "739d372726cb23129d57a539ce1432453448816e345e1545f6127296926b6754 /tmp/cmake-install.sh" | sha256sum --check \
       && chmod u+x /tmp/cmake-install.sh \
       && /tmp/cmake-install.sh --skip-license --prefix=/usr/local/ \
       && rm /tmp/cmake-install.sh
@@ -525,25 +515,6 @@ RUN wget https://github.com/ChenHuajun/pg_roaringbitmap/archive/refs/tags/v0.5.4
     make -j $(getconf _NPROCESSORS_ONLN) install && \
     echo 'trusted = true' >> /usr/local/pgsql/share/extension/roaringbitmap.control
 
-#########################################################################################
-#
-# Layer "pg-embedding-pg-build"
-# compile pg_embedding extension
-#
-#########################################################################################
-FROM build-deps AS pg-embedding-pg-build
-COPY --from=pg-build /usr/local/pgsql/ /usr/local/pgsql/
-
-ENV PATH "/usr/local/pgsql/bin/:$PATH"
-# 2465f831ea1f8d49c1d74f8959adb7fc277d70cd made on 05/07/2023
-# There is no release tag yet
-RUN wget https://github.com/neondatabase/pg_embedding/archive/2465f831ea1f8d49c1d74f8959adb7fc277d70cd.tar.gz -O pg_embedding.tar.gz && \
-    echo "047af2b1f664a1e6e37867bd4eeaf5934fa27d6ba3d6c4461efa388ddf7cd1d5 pg_embedding.tar.gz" | sha256sum --check && \
-    mkdir pg_embedding-src && cd pg_embedding-src && tar xvzf ../pg_embedding.tar.gz --strip-components=1 -C . && \
-    make -j $(getconf _NPROCESSORS_ONLN) && \
-    make -j $(getconf _NPROCESSORS_ONLN) install && \
-    echo 'trusted = true' >> /usr/local/pgsql/share/extension/embedding.control
-
 #########################################################################################
 #
 # Layer "pg-anon-pg-build"
@@ -700,7 +671,6 @@ COPY --from=pg-pgx-ulid-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=rdkit-pg-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg-uuidv7-pg-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg-roaringbitmap-pg-build /usr/local/pgsql/ /usr/local/pgsql/
-COPY --from=pg-embedding-pg-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY pgxn/ pgxn/
 
 RUN make -j $(getconf _NPROCESSORS_ONLN) \

Makefile

@@ -83,8 +83,6 @@ $(POSTGRES_INSTALL_DIR)/build/%/config.status:
 # I'm not sure why it wouldn't work, but this is the only place (apart from
 # the "build-all-versions" entry points) where direct mention of PostgreSQL
 # versions is used.
-.PHONY: postgres-configure-v16
-postgres-configure-v16: $(POSTGRES_INSTALL_DIR)/build/v16/config.status
 .PHONY: postgres-configure-v15
 postgres-configure-v15: $(POSTGRES_INSTALL_DIR)/build/v15/config.status
 .PHONY: postgres-configure-v14
@@ -167,33 +165,28 @@ neon-pg-ext-clean-%:
 .PHONY: neon-pg-ext
 neon-pg-ext: \
 	neon-pg-ext-v14 \
-	neon-pg-ext-v15 \
-	neon-pg-ext-v16
+	neon-pg-ext-v15
 
 .PHONY: neon-pg-ext-clean
 neon-pg-ext-clean: \
 	neon-pg-ext-clean-v14 \
-	neon-pg-ext-clean-v15 \
-	neon-pg-ext-clean-v16
+	neon-pg-ext-clean-v15
 
 # shorthand to build all Postgres versions
 .PHONY: postgres
 postgres: \
 	postgres-v14 \
-	postgres-v15 \
-	postgres-v16
+	postgres-v15
 
 .PHONY: postgres-headers
 postgres-headers: \
 	postgres-headers-v14 \
-	postgres-headers-v15 \
-	postgres-headers-v16
+	postgres-headers-v15
 
 .PHONY: postgres-clean
 postgres-clean: \
 	postgres-clean-v14 \
-	postgres-clean-v15 \
-	postgres-clean-v16
+	postgres-clean-v15
 
 # This doesn't remove the effects of 'configure'.
 .PHONY: clean

compute_tools/Cargo.toml

@@ -6,10 +6,8 @@ license.workspace = true
 
 [dependencies]
 anyhow.workspace = true
-async-compression.workspace = true
 chrono.workspace = true
 clap.workspace = true
-flate2.workspace = true
 futures.workspace = true
 hyper = { workspace = true, features = ["full"] }
 notify.workspace = true

compute_tools/src/compute.rs

@@ -1,5 +1,4 @@
 use std::fs;
-use std::io::BufRead;
 use std::os::unix::fs::PermissionsExt;
 use std::path::Path;
 use std::process::{Command, Stdio};
@@ -16,7 +15,6 @@ use utils::lsn::Lsn;
 
 use compute_api::responses::{ComputeMetrics, ComputeStatus};
 use compute_api::spec::{ComputeMode, ComputeSpec};
-use utils::measured_stream::MeasuredReader;
 
 use crate::config;
 use crate::pg_helpers::*;
@@ -142,14 +140,14 @@ fn create_neon_superuser(spec: &ComputeSpec, client: &mut Client) -> Result<()>
         .cluster
         .roles
         .iter()
-        .map(|r| escape_literal(&r.name))
+        .map(|r| format!("'{}'", escape_literal(&r.name)))
         .collect::<Vec<_>>();
 
     let dbs = spec
         .cluster
         .databases
         .iter()
-        .map(|db| escape_literal(&db.name))
+        .map(|db| format!("'{}'", escape_literal(&db.name)))
         .collect::<Vec<_>>();
 
     let roles_decl = if roles.is_empty() {
@@ -255,52 +253,20 @@ impl ComputeNode {
 
         let mut client = config.connect(NoTls)?;
         let basebackup_cmd = match lsn {
-            // HACK We don't use compression on first start (Lsn(0)) because there's no API for it
-            Lsn(0) => format!("basebackup {} {}", spec.tenant_id, spec.timeline_id),
-            _ => format!(
-                "basebackup {} {} {} --gzip",
-                spec.tenant_id, spec.timeline_id, lsn
-            ),
+            Lsn(0) => format!("basebackup {} {}", spec.tenant_id, spec.timeline_id), // First start of the compute
+            _ => format!("basebackup {} {} {}", spec.tenant_id, spec.timeline_id, lsn),
         };
-
         let copyreader = client.copy_out(basebackup_cmd.as_str())?;
-        let mut measured_reader = MeasuredReader::new(copyreader);
-
-        // Check the magic number to see if it's a gzip or not. Even though
-        // we might explicitly ask for gzip, an old pageserver with no implementation
-        // of gzip compression might send us uncompressed data. After some time
-        // passes we can assume all pageservers know how to compress and we can
-        // delete this check.
-        //
-        // If the data is not gzip, it will be tar. It will not be mistakenly
-        // recognized as gzip because tar starts with an ascii encoding of a filename,
-        // and 0x1f and 0x8b are unlikely first characters for any filename. Moreover,
-        // we send the "global" directory first from the pageserver, so it definitely
-        // won't be recognized as gzip.
-        let mut bufreader = std::io::BufReader::new(&mut measured_reader);
-        let gzip = {
-            let peek = bufreader.fill_buf().unwrap();
-            peek[0] == 0x1f && peek[1] == 0x8b
-        };
 
         // Read the archive directly from the `CopyOutReader`
         //
         // Set `ignore_zeros` so that unpack() reads all the Copy data and
         // doesn't stop at the end-of-archive marker. Otherwise, if the server
         // sends an Error after finishing the tarball, we will not notice it.
-        if gzip {
-            let mut ar = tar::Archive::new(flate2::read::GzDecoder::new(&mut bufreader));
-            ar.set_ignore_zeros(true);
-            ar.unpack(&self.pgdata)?;
-        } else {
-            let mut ar = tar::Archive::new(&mut bufreader);
-            ar.set_ignore_zeros(true);
-            ar.unpack(&self.pgdata)?;
-        };
+        let mut ar = tar::Archive::new(copyreader);
+        ar.set_ignore_zeros(true);
+        ar.unpack(&self.pgdata)?;
 
-        // Report metrics
-        self.state.lock().unwrap().metrics.basebackup_bytes =
-            measured_reader.get_byte_count() as u64;
         self.state.lock().unwrap().metrics.basebackup_ms = Utc::now()
             .signed_duration_since(start_time)
             .to_std()
@@ -583,13 +549,6 @@ impl ComputeNode {
             pspec.spec.cluster.cluster_id.as_deref().unwrap_or("None")
         );
 
-        // Log metrics so that we can search for slow operations in logs
-        let metrics = {
-            let state = self.state.lock().unwrap();
-            state.metrics.clone()
-        };
-        info!(?metrics, "compute start finished");
-
         Ok(pg)
     }
 

compute_tools/src/config.rs

@@ -47,22 +47,30 @@ pub fn write_postgres_conf(path: &Path, spec: &ComputeSpec) -> Result<()> {
     // Add options for connecting to storage
     writeln!(file, "# Neon storage settings")?;
     if let Some(s) = &spec.pageserver_connstring {
-        writeln!(file, "neon.pageserver_connstring={}", escape_conf_value(s))?;
+        writeln!(
+            file,
+            "neon.pageserver_connstring='{}'",
+            escape_conf_value(s)
+        )?;
     }
     if !spec.safekeeper_connstrings.is_empty() {
         writeln!(
             file,
-            "neon.safekeepers={}",
+            "neon.safekeepers='{}'",
             escape_conf_value(&spec.safekeeper_connstrings.join(","))
         )?;
     }
     if let Some(s) = &spec.tenant_id {
-        writeln!(file, "neon.tenant_id={}", escape_conf_value(&s.to_string()))?;
+        writeln!(
+            file,
+            "neon.tenant_id='{}'",
+            escape_conf_value(&s.to_string())
+        )?;
     }
     if let Some(s) = &spec.timeline_id {
         writeln!(
             file,
-            "neon.timeline_id={}",
+            "neon.timeline_id='{}'",
             escape_conf_value(&s.to_string())
         )?;
     }

compute_tools/src/pg_helpers.rs

@@ -16,26 +16,15 @@ use compute_api::spec::{Database, GenericOption, GenericOptions, PgIdent, Role};
 
 const POSTGRES_WAIT_TIMEOUT: Duration = Duration::from_millis(60 * 1000); // milliseconds
 
-/// Escape a string for including it in a SQL literal. Wrapping the result
-/// with `E'{}'` or `'{}'` is not required, as it returns a ready-to-use
-/// SQL string literal, e.g. `'db'''` or `E'db\\'`.
-/// See <https://github.com/postgres/postgres/blob/da98d005cdbcd45af563d0c4ac86d0e9772cd15f/src/backend/utils/adt/quote.c#L47>
-/// for the original implementation.
+/// Escape a string for including it in a SQL literal
 pub fn escape_literal(s: &str) -> String {
-    let res = s.replace('\'', "''").replace('\\', "\\\\");
-
-    if res.contains('\\') {
-        format!("E'{}'", res)
-    } else {
-        format!("'{}'", res)
-    }
+    s.replace('\'', "''").replace('\\', "\\\\")
 }
 
-/// Escape a string so that it can be used in postgresql.conf. Wrapping the result
-/// with `'{}'` is not required, as it returns a ready-to-use config string.
+/// Escape a string so that it can be used in postgresql.conf.
+/// Same as escape_literal, currently.
 pub fn escape_conf_value(s: &str) -> String {
-    let res = s.replace('\'', "''").replace('\\', "\\\\");
-    format!("'{}'", res)
+    s.replace('\'', "''").replace('\\', "\\\\")
 }
 
 trait GenericOptionExt {
@@ -48,7 +37,7 @@ impl GenericOptionExt for GenericOption {
     fn to_pg_option(&self) -> String {
         if let Some(val) = &self.value {
             match self.vartype.as_ref() {
-                "string" => format!("{} {}", self.name, escape_literal(val)),
+                "string" => format!("{} '{}'", self.name, escape_literal(val)),
                 _ => format!("{} {}", self.name, val),
             }
         } else {
@@ -60,7 +49,7 @@ impl GenericOptionExt for GenericOption {
     fn to_pg_setting(&self) -> String {
         if let Some(val) = &self.value {
             match self.vartype.as_ref() {
-                "string" => format!("{} = {}", self.name, escape_conf_value(val)),
+                "string" => format!("{} = '{}'", self.name, escape_conf_value(val)),
                 _ => format!("{} = {}", self.name, val),
             }
         } else {

compute_tools/src/spec.rs

@@ -397,44 +397,10 @@ pub fn handle_databases(spec: &ComputeSpec, client: &mut Client) -> Result<()> {
                 // We do not check either DB exists or not,
                 // Postgres will take care of it for us
                 "delete_db" => {
-                    // In Postgres we can't drop a database if it is a template.
-                    // So we need to unset the template flag first, but it could
-                    // be a retry, so we could've already dropped the database.
-                    // Check that database exists first to make it idempotent.
-                    let unset_template_query: String = format!(
-                        "
-                        DO $$
-                        BEGIN
-                            IF EXISTS(
-                                SELECT 1
-                                FROM pg_catalog.pg_database
-                                WHERE datname = {}
-                            )
-                            THEN
-                            ALTER DATABASE {} is_template false;
-                            END IF;
-                        END
-                        $$;",
-                        escape_literal(&op.name),
-                        &op.name.pg_quote()
-                    );
-                    // Use FORCE to drop database even if there are active connections.
-                    // We run this from `cloud_admin`, so it should have enough privileges.
-                    // NB: there could be other db states, which prevent us from dropping
-                    // the database. For example, if db is used by any active subscription
-                    // or replication slot.
-                    // TODO: deal with it once we allow logical replication. Proper fix should
-                    // involve returning an error code to the control plane, so it could
-                    // figure out that this is a non-retryable error, return it to the user
-                    // and fail operation permanently.
-                    let drop_db_query: String = format!(
-                        "DROP DATABASE IF EXISTS {} WITH (FORCE)",
-                        &op.name.pg_quote()
-                    );
+                    let query: String = format!("DROP DATABASE IF EXISTS {}", &op.name.pg_quote());
 
                     warn!("deleting database '{}'", &op.name);
-                    client.execute(unset_template_query.as_str(), &[])?;
-                    client.execute(drop_db_query.as_str(), &[])?;
+                    client.execute(query.as_str(), &[])?;
                 }
                 "rename_db" => {
                     let new_name = op.new_name.as_ref().unwrap();

compute_tools/tests/pg_helpers_tests.rs

@@ -89,12 +89,4 @@ test.escaping = 'here''s a backslash \\ and a quote '' and a double-quote " hoor
         assert_eq!(none_generic_options.find("missed_value"), None);
         assert_eq!(none_generic_options.find("invalid_value"), None);
     }
-
-    #[test]
-    fn test_escape_literal() {
-        assert_eq!(escape_literal("test"), "'test'");
-        assert_eq!(escape_literal("test'"), "'test'''");
-        assert_eq!(escape_literal("test\\'"), "E'test\\\\'''");
-        assert_eq!(escape_literal("test\\'\\'"), "E'test\\\\''\\\\'''");
-    }
 }

control_plane/src/background_process.rs

@@ -10,7 +10,7 @@
 //! (non-Neon binaries don't necessarily follow our pidfile conventions).
 //! The pid stored in the file is later used to stop the service.
 //!
-//! See the [`lock_file`](utils::lock_file) module for more info.
+//! See [`lock_file`] module for more info.
 
 use std::ffi::OsStr;
 use std::io::Write;

control_plane/src/broker.rs

@@ -2,9 +2,8 @@
 //!
 //! In the local test environment, the data for each safekeeper is stored in
 //!
-//! ```text
 //!   .neon/safekeepers/<safekeeper id>
-//! ```
+//!
 use anyhow::Context;
 
 use std::path::PathBuf;

control_plane/src/endpoint.rs

@@ -2,9 +2,7 @@
 //!
 //! In the local test environment, the data for each endpoint is stored in
 //!
-//! ```text
 //!   .neon/endpoints/<endpoint id>
-//! ```
 //!
 //! Some basic information about the endpoint, like the tenant and timeline IDs,
 //! are stored in the `endpoint.json` file. The `endpoint.json` file is created
@@ -24,7 +22,7 @@
 //!
 //! Directory contents:
 //!
-//! ```text
+//! ```ignore
 //! .neon/endpoints/main/
 //!     compute.log               - log output of `compute_ctl` and `postgres`
 //!     endpoint.json             - serialized `EndpointConf` struct

control_plane/src/local_env.rs

@@ -169,7 +169,6 @@ impl LocalEnv {
         match pg_version {
             14 => Ok(path.join(format!("v{pg_version}"))),
             15 => Ok(path.join(format!("v{pg_version}"))),
-            16 => Ok(path.join(format!("v{pg_version}"))),
             _ => bail!("Unsupported postgres version: {}", pg_version),
         }
     }
@@ -178,7 +177,6 @@ impl LocalEnv {
         match pg_version {
             14 => Ok(self.pg_distrib_dir(pg_version)?.join("bin")),
             15 => Ok(self.pg_distrib_dir(pg_version)?.join("bin")),
-            16 => Ok(self.pg_distrib_dir(pg_version)?.join("bin")),
             _ => bail!("Unsupported postgres version: {}", pg_version),
         }
     }
@@ -186,7 +184,6 @@ impl LocalEnv {
         match pg_version {
             14 => Ok(self.pg_distrib_dir(pg_version)?.join("lib")),
             15 => Ok(self.pg_distrib_dir(pg_version)?.join("lib")),
-            16 => Ok(self.pg_distrib_dir(pg_version)?.join("lib")),
             _ => bail!("Unsupported postgres version: {}", pg_version),
         }
     }

control_plane/src/safekeeper.rs

@@ -2,9 +2,8 @@
 //!
 //! In the local test environment, the data for each safekeeper is stored in
 //!
-//! ```text
 //!   .neon/safekeepers/<safekeeper id>
-//! ```
+//!
 use std::io::Write;
 use std::path::PathBuf;
 use std::process::Child;

docker-compose/docker-compose.yml

@@ -189,7 +189,7 @@ services:
       - "/bin/bash"
       - "-c"
     command:
-      - "until pg_isready -h compute -p 55433 -U cloud_admin ; do
+      - "until pg_isready -h compute -p 55433 ; do
             echo 'Waiting to start compute...' && sleep 1;
          done"
     depends_on:

docs/docker.md

@@ -48,7 +48,6 @@ Creating docker-compose_storage_broker_1       ... done
 2. connect compute node
 ```
 $ echo "localhost:55433:postgres:cloud_admin:cloud_admin" >> ~/.pgpass
-$ chmod 600 ~/.pgpass
 $ psql -h localhost -p 55433 -U cloud_admin
 postgres=# CREATE TABLE t(key int primary key, value text);
 CREATE TABLE

docs/rfcs/024-user-mgmt.md

@@ -1,84 +0,0 @@
-# Postgres user and database management
-
-(This supersedes the previous proposal that looked too complicated and desynchronization-prone)
-
-We've accumulated a bunch of problems with our approach to role and database management, namely:
-
-1. we don't allow role and database creation from Postgres, and users are complaining about that
-2. fine-grained role management is not possible both from Postgres and console
-
-Right now, we do store users and databases both in console and Postgres, and there are two main reasons for
-that:
-
-* we want to be able to authenticate users in proxy against the console without Postgres' involvement. Otherwise,
-malicious brute force attempts will wake up Postgres (expensive) and may exhaust the Postgres connections limit (deny of service).
-* it is handy when we can render console UI without waking up compute (e.g., show database list)
-
-This RFC doesn't talk about giving root access to the database, which is blocked by a secure runtime setup.
-
-## Overview
-
-* Add Postgres extension that sends an HTTP request each time transaction that modifies users/databases is about to commit.
-* Add user management API to internal console API. Also, the console should put a JWT token into the compute so that it can access management API.
-
-## Postgres behavior
-
-The default user role (@username) should have `CREATE ROLE`, `CREATE DB`, and `BYPASSRLS` privileges. We expose the Postgres port
-to the open internet, so we need to check password strength. Now console generates strong passwords, so there is no risk of having dumb passwords. With user-provided passwords, such risks exist.
-
-Since we store passwords in the console we should also send unencrypted password when role is created/changed. Hence communication with the console must be encrypted. Postgres also supports creating roles using hashes, in that case, we will not be able to get a raw password. So I can see the following options here:
-  * roles created via SQL will *not* have raw passwords in the console
-  * roles created via SQL will have raw passwords in the console, except ones that were created using hashes
-
-I'm leaning towards the second option here as it is a bit more consistent one -- if raw password storage is enabled then we store passwords in all cases where we can store them.
-
-To send data about roles and databases from Postgres to the console we can create the following Postgres extension:
-
-  * Intercept role/database changes in `ProcessUtility_hook`. Here we have access to the query statement with the raw password. The hook handler itself should not dial the console immediately and rather stash info in some hashmap for later use.
-  * When the transaction is about to commit we execute collected role modifications (all as one -- console should either accept all or reject all, and hence API shouldn't be REST-like). If the console request fails we can roll back the transaction. This way if the transaction is committed we know for sure that console has this information. We can use `XACT_EVENT_PRE_COMMIT` and `XACT_EVENT_PARALLEL_PRE_COMMIT` for that.
-  * Extension should be mindful of the fact that it is possible to create and delete roles within the transaction.
-  * We also need to track who is database owner, some coding around may be needed to get the current user when the database is created.
-
-## Console user management API
-
-The current public API has REST API for role management. We need to have some analog for the internal API (called mgmt API in the console code). But unlike public API here we want to have an atomic way to create several roles/databases (in cases when several roles were created in the same transaction). So something like that may work:
-
-```
-curl -X PATCH /api/v1/roles_and_databases -d '
-[
-    {"op":"create", "type":"role", "name": "kurt", "password":"lYgT3BlbkFJ2vBZrqv"},
-    {"op":"drop", "type":"role", "name": "trout"},
-    {"op":"alter", "type":"role", "name": "kilgore", "password":"3BlbkFJ2vB"},
-    {"op":"create", "type":"database", "name": "db2", "owner": "eliot"},
-]
-'
-```
-
-Makes sense not to error out on duplicated create/delete operations (see failure modes)
-
-## Managing users from the console
-
-Now console puts a spec file with the list of databases/roles and delta operations in all the compute pods. `compute_ctl` then picks up that file and stubbornly executes deltas and checks data in the spec file is the same as in the Postgres. This way if the user creates a role in the UI we restart compute with a new spec file and during the start databases/roles are created. So if Postgres send an HTTP call each time role is created we need to break recursion in that case. We can do that based on application_name or some GUC or user (local == no HTTP hook).
-
-Generally, we have several options when we are creating users via console:
-
-1. restart compute with a new spec file, execute local SQL command; cut recursion in the extension
-2. "push" spec files into running compute, execute local SQL command; cut recursion in the extension
-3. "push" spec files into running compute, execute local SQL command; let extension create those roles in the console
-4. avoid managing roles via spec files, send SQL commands to compute; let extension create those roles in the console
-
-The last option is the most straightforward one, but with the raw password storage opt-out, we will not have the password to establish an SQL connection. Also, we need a spec for provisioning purposes and to address potential desync (but that is quite unlikely). So I think the easiest approach would be:
-
-1. keep role management like it is now and cut the recursion in the extension when SQL is executed by compute_ctl
-2. add "push" endpoint to the compute_ctl to avoid compute restart during the `apply_config` operation -- that can be done as a follow up to avoid increasing scope too much
-
-## Failure modes
-
-* during role creation via SQL role was created in the console but the connection was dropped before Postgres got acknowledgment or some error happened after acknowledgment (out of disk space, deadlock, etc):
-
-  in that case, Postgres won't have a role that exists in the console. Compute restart will heal it (due to the spec file). Also if the console allows repeated creation/deletion user can repeat the transaction.
-
-
-# Scalability
-
-On my laptop, I can create 4200 roles per second. That corresponds to 363 million roles per day. Since each role creation ends up in the console database we can add some limit to the number of roles (could be reasonably big to not run into it often -- like 1k or 10k).

docs/tools.md

@@ -1,22 +0,0 @@
-# Useful development tools
-
-This readme contains some hints on how to set up some optional development tools.
-
-## ccls
-
-[ccls](https://github.com/MaskRay/ccls) is a c/c++ language server. It requires some setup
-to work well. There are different ways to do it but here's what works for me:
-1. Make a common parent directory for all your common neon projects. (for example, `~/src/neondatabase/`)
-2. Go to `vendor/postgres-v15`
-3. Run `make clean && ./configure`
-4. Install [bear](https://github.com/rizsotto/Bear), and run `bear -- make -j4`
-5. Copy the generated `compile_commands.json` to `~/src/neondatabase` (or equivalent)
-6. Run `touch ~/src/neondatabase/.ccls-root` this will make the `compile_commands.json` file discoverable in all subdirectories
-
-With this setup you will get decent lsp mileage inside the postgres repo, and also any postgres extensions that you put in `~/src/neondatabase/`, like `pg_embedding`, or inside `~/src/neondatabase/neon/pgxn` as well.
-
-Some additional tips for various IDEs:
-
-### Emacs
-
-To improve performance: `(setq lsp-lens-enable nil)`

libs/compute_api/src/responses.rs

@@ -71,7 +71,6 @@ pub struct ComputeMetrics {
     pub wait_for_spec_ms: u64,
     pub sync_safekeepers_ms: u64,
     pub basebackup_ms: u64,
-    pub basebackup_bytes: u64,
     pub start_postgres_ms: u64,
     pub config_ms: u64,
     pub total_startup_ms: u64,

libs/metrics/src/metric_vec_duration.rs

@@ -1,4 +1,4 @@
-//! Helpers for observing duration on `HistogramVec` / `CounterVec` / `GaugeVec` / `MetricVec<T>`.
+//! Helpers for observing duration on HistogramVec / CounterVec / GaugeVec / MetricVec<T>.
 
 use std::{future::Future, time::Instant};
 

libs/pageserver_api/src/models.rs

@@ -411,16 +411,12 @@ pub struct LayerResidenceEvent {
     pub reason: LayerResidenceEventReason,
 }
 
-/// The reason for recording a given [`LayerResidenceEvent`].
+/// The reason for recording a given [`ResidenceEvent`].
 #[derive(Debug, Clone, Copy, Serialize, Deserialize)]
 pub enum LayerResidenceEventReason {
     /// The layer map is being populated, e.g. during timeline load or attach.
     /// This includes [`RemoteLayer`] objects created in [`reconcile_with_remote`].
     /// We need to record such events because there is no persistent storage for the events.
-    ///
-    // https://github.com/rust-lang/rust/issues/74481
-    /// [`RemoteLayer`]: ../../tenant/storage_layer/struct.RemoteLayer.html
-    /// [`reconcile_with_remote`]: ../../tenant/struct.Timeline.html#method.reconcile_with_remote
     LayerLoad,
     /// We just created the layer (e.g., freeze_and_flush or compaction).
     /// Such layers are always [`LayerResidenceStatus::Resident`].

libs/pageserver_api/src/reltag.rs

@@ -60,9 +60,8 @@ impl Ord for RelTag {
 
 /// Display RelTag in the same format that's used in most PostgreSQL debug messages:
 ///
-/// ```text
 /// <spcnode>/<dbnode>/<relnode>[_fsm|_vm|_init]
-/// ```
+///
 impl fmt::Display for RelTag {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         if let Some(forkname) = forknumber_to_name(self.forknum) {

libs/postgres_ffi/build.rs

@@ -56,7 +56,7 @@ fn main() -> anyhow::Result<()> {
         PathBuf::from("pg_install")
     };
 
-    for pg_version in &["v14", "v15", "v16"] {
+    for pg_version in &["v14", "v15"] {
         let mut pg_install_dir_versioned = pg_install_dir.join(pg_version);
         if pg_install_dir_versioned.is_relative() {
             let cwd = env::current_dir().context("Failed to get current_dir")?;

libs/postgres_ffi/src/lib.rs

@@ -51,7 +51,6 @@ macro_rules! for_all_postgres_versions {
     ($macro:tt) => {
         $macro!(v14);
         $macro!(v15);
-        $macro!(v16);
     };
 }
 
@@ -93,10 +92,9 @@ pub use v14::bindings::DBState_DB_SHUTDOWNED;
 pub fn bkpimage_is_compressed(bimg_info: u8, version: u32) -> anyhow::Result<bool> {
     match version {
         14 => Ok(bimg_info & v14::bindings::BKPIMAGE_IS_COMPRESSED != 0),
-        15 | 16 => Ok(bimg_info & v15::bindings::BKPIMAGE_COMPRESS_PGLZ != 0
+        15 => Ok(bimg_info & v15::bindings::BKPIMAGE_COMPRESS_PGLZ != 0
             || bimg_info & v15::bindings::BKPIMAGE_COMPRESS_LZ4 != 0
             || bimg_info & v15::bindings::BKPIMAGE_COMPRESS_ZSTD != 0),
-
         _ => anyhow::bail!("Unknown version {}", version),
     }
 }
@@ -112,7 +110,6 @@ pub fn generate_wal_segment(
     match pg_version {
         14 => v14::xlog_utils::generate_wal_segment(segno, system_id, lsn),
         15 => v15::xlog_utils::generate_wal_segment(segno, system_id, lsn),
-        16 => v16::xlog_utils::generate_wal_segment(segno, system_id, lsn),
         _ => Err(SerializeError::BadInput),
     }
 }
@@ -126,7 +123,6 @@ pub fn generate_pg_control(
     match pg_version {
         14 => v14::xlog_utils::generate_pg_control(pg_control_bytes, checkpoint_bytes, lsn),
         15 => v15::xlog_utils::generate_pg_control(pg_control_bytes, checkpoint_bytes, lsn),
-        16 => v16::xlog_utils::generate_pg_control(pg_control_bytes, checkpoint_bytes, lsn),
         _ => anyhow::bail!("Unknown version {}", pg_version),
     }
 }
@@ -201,7 +197,7 @@ pub fn fsm_logical_to_physical(addr: BlockNumber) -> BlockNumber {
 
 pub mod waldecoder {
 
-    use crate::{v14, v15, v16};
+    use crate::{v14, v15};
     use bytes::{Buf, Bytes, BytesMut};
     use std::num::NonZeroU32;
     use thiserror::Error;
@@ -263,10 +259,6 @@ pub mod waldecoder {
                     use self::v15::waldecoder_handler::WalStreamDecoderHandler;
                     self.poll_decode_internal()
                 }
-                16 => {
-                    use self::v16::waldecoder_handler::WalStreamDecoderHandler;
-                    self.poll_decode_internal()
-                }
                 _ => Err(WalDecodeError {
                     msg: format!("Unknown version {}", self.pg_version),
                     lsn: self.lsn,

libs/postgres_ffi/src/pg_constants_v16.rs

@@ -1 +0,0 @@
-

libs/postgres_ffi/src/relfile_utils.rs

@@ -49,16 +49,14 @@ pub fn forknumber_to_name(forknum: u8) -> Option<&'static str> {
     }
 }
 
+///
 /// Parse a filename of a relation file. Returns (relfilenode, forknum, segno) tuple.
 ///
 /// Formats:
-///
-/// ```text
 /// <oid>
 /// <oid>_<fork name>
 /// <oid>.<segment number>
 /// <oid>_<fork name>.<segment number>
-/// ```
 ///
 /// See functions relpath() and _mdfd_segpath() in PostgreSQL sources.
 ///

libs/postgres_ffi/wal_craft/src/lib.rs

@@ -52,7 +52,6 @@ impl Conf {
         match self.pg_version {
             14 => Ok(path.join(format!("v{}", self.pg_version))),
             15 => Ok(path.join(format!("v{}", self.pg_version))),
-            16 => Ok(path.join(format!("v{}", self.pg_version))),
             _ => bail!("Unsupported postgres version: {}", self.pg_version),
         }
     }

libs/pq_proto/src/framed.rs

@@ -5,11 +5,11 @@
 //! It is similar to what tokio_util::codec::Framed with appropriate codec
 //! provides, but `FramedReader` and `FramedWriter` read/write parts can be used
 //! separately without using split from futures::stream::StreamExt (which
-//! allocates a [Box] in polling internally). tokio::io::split is used for splitting
+//! allocates box[1] in polling internally). tokio::io::split is used for splitting
 //! instead. Plus we customize error messages more than a single type for all io
 //! calls.
 //!
-//! [Box]: https://docs.rs/futures-util/0.3.26/src/futures_util/lock/bilock.rs.html#107
+//! [1] https://docs.rs/futures-util/0.3.26/src/futures_util/lock/bilock.rs.html#107
 use bytes::{Buf, BytesMut};
 use std::{
     future::Future,
@@ -117,7 +117,7 @@ impl<S: AsyncWrite + Unpin> Framed<S> {
 impl<S: AsyncRead + AsyncWrite + Unpin> Framed<S> {
     /// Split into owned read and write parts. Beware of potential issues with
     /// using halves in different tasks on TLS stream:
-    /// <https://github.com/tokio-rs/tls/issues/40>
+    /// https://github.com/tokio-rs/tls/issues/40
     pub fn split(self) -> (FramedReader<S>, FramedWriter<S>) {
         let (read_half, write_half) = tokio::io::split(self.stream);
         let reader = FramedReader {

libs/pq_proto/src/lib.rs

@@ -934,15 +934,6 @@ impl<'a> BeMessage<'a> {
     }
 }
 
-fn terminate_code(code: &[u8; 5]) -> [u8; 6] {
-    let mut terminated = [0; 6];
-    for (i, &elem) in code.iter().enumerate() {
-        terminated[i] = elem;
-    }
-
-    terminated
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -974,3 +965,12 @@ mod tests {
         assert_eq!(split_options(&params), ["foo bar", " \\", "baz ", "lol"]);
     }
 }
+
+fn terminate_code(code: &[u8; 5]) -> [u8; 6] {
+    let mut terminated = [0; 6];
+    for (i, &elem) in code.iter().enumerate() {
+        terminated[i] = elem;
+    }
+
+    terminated
+}

libs/remote_storage/src/lib.rs

@@ -34,12 +34,12 @@ pub const DEFAULT_REMOTE_STORAGE_MAX_CONCURRENT_SYNCS: usize = 50;
 pub const DEFAULT_REMOTE_STORAGE_MAX_SYNC_ERRORS: u32 = 10;
 /// Currently, sync happens with AWS S3, that has two limits on requests per second:
 /// ~200 RPS for IAM services
-/// <https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/UsingWithRDS.IAMDBAuth.html>
+/// https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/UsingWithRDS.IAMDBAuth.html
 /// ~3500 PUT/COPY/POST/DELETE or 5500 GET/HEAD S3 requests
-/// <https://aws.amazon.com/premiumsupport/knowledge-center/s3-request-limit-avoid-throttling/>
+/// https://aws.amazon.com/premiumsupport/knowledge-center/s3-request-limit-avoid-throttling/
 pub const DEFAULT_REMOTE_STORAGE_S3_CONCURRENCY_LIMIT: usize = 100;
 /// No limits on the client side, which currenltly means 1000 for AWS S3.
-/// <https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListObjectsV2.html#API_ListObjectsV2_RequestSyntax>
+/// https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListObjectsV2.html#API_ListObjectsV2_RequestSyntax
 pub const DEFAULT_MAX_KEYS_PER_LIST_RESPONSE: Option<i32> = None;
 
 const REMOTE_STORAGE_PREFIX_SEPARATOR: char = '/';
@@ -50,12 +50,6 @@ const REMOTE_STORAGE_PREFIX_SEPARATOR: char = '/';
 #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
 pub struct RemotePath(PathBuf);
 
-impl std::fmt::Display for RemotePath {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{}", self.0.display())
-    }
-}
-
 impl RemotePath {
     pub fn new(relative_path: &Path) -> anyhow::Result<Self> {
         anyhow::ensure!(

libs/remote_storage/src/local_fs.rs

@@ -7,7 +7,6 @@
 use std::{
     borrow::Cow,
     future::Future,
-    io::ErrorKind,
     path::{Path, PathBuf},
     pin::Pin,
 };
@@ -151,7 +150,10 @@ impl RemoteStorage for LocalFs {
         let mut files = vec![];
         let mut directory_queue = vec![full_path.clone()];
 
-        while let Some(cur_folder) = directory_queue.pop() {
+        while !directory_queue.is_empty() {
+            let cur_folder = directory_queue
+                .pop()
+                .expect("queue cannot be empty: we just checked");
             let mut entries = fs::read_dir(cur_folder.clone()).await?;
             while let Some(entry) = entries.next_entry().await? {
                 let file_name: PathBuf = entry.file_name().into();
@@ -341,14 +343,18 @@ impl RemoteStorage for LocalFs {
 
     async fn delete(&self, path: &RemotePath) -> anyhow::Result<()> {
         let file_path = path.with_base(&self.storage_root);
-        match fs::remove_file(&file_path).await {
-            Ok(()) => Ok(()),
-            // The file doesn't exist. This shouldn't yield an error to mirror S3's behaviour.
+        if !file_path.exists() {
             // See https://docs.aws.amazon.com/AmazonS3/latest/API/API_DeleteObject.html
             // > If there isn't a null version, Amazon S3 does not remove any objects but will still respond that the command was successful.
-            Err(e) if e.kind() == ErrorKind::NotFound => Ok(()),
-            Err(e) => Err(anyhow::anyhow!(e)),
+            return Ok(());
         }
+
+        if !file_path.is_file() {
+            anyhow::bail!("{file_path:?} is not a file");
+        }
+        Ok(fs::remove_file(file_path)
+            .await
+            .map_err(|e| anyhow::anyhow!(e))?)
     }
 
     async fn delete_objects<'a>(&self, paths: &'a [RemotePath]) -> anyhow::Result<()> {

libs/tenant_size_model/src/calculation.rs

@@ -21,7 +21,7 @@ use crate::{SegmentMethod, SegmentSizeResult, SizeResult, StorageModel};
 // 2. D+C+a+b
 // 3. D+A+B
 
-/// `Segment` which has had its size calculated.
+/// [`Segment`] which has had it's size calculated.
 #[derive(Clone, Debug)]
 struct SegmentSize {
     method: SegmentMethod,

libs/tracing-utils/src/http.rs

@@ -33,7 +33,7 @@ pub enum OtelName<'a> {
 /// directly into HTTP servers. However, I couldn't find one for Hyper,
 /// so I had to write our own. OpenTelemetry website has a registry of
 /// instrumentation libraries at:
-/// <https://opentelemetry.io/registry/?language=rust&component=instrumentation>
+/// https://opentelemetry.io/registry/?language=rust&component=instrumentation
 /// If a Hyper crate appears, consider switching to that.
 pub async fn tracing_handler<F, R>(
     req: Request<Body>,

libs/utils/Cargo.toml

@@ -40,8 +40,6 @@ pq_proto.workspace = true
 metrics.workspace = true
 workspace_hack.workspace = true
 
-const_format.workspace = true
-
 [dev-dependencies]
 byteorder.workspace = true
 bytes.workspace = true

libs/utils/src/http/json.rs

@@ -14,7 +14,7 @@ pub async fn json_request<T: for<'de> Deserialize<'de>>(
         .map_err(ApiError::BadRequest)
 }
 
-/// Will be removed as part of <https://github.com/neondatabase/neon/issues/4282>
+/// Will be removed as part of https://github.com/neondatabase/neon/issues/4282
 pub async fn json_request_or_empty_body<T: for<'de> Deserialize<'de>>(
     request: &mut Request<Body>,
 ) -> Result<Option<T>, ApiError> {

libs/utils/src/lib.rs

@@ -109,16 +109,10 @@ pub use failpoint_macro_helpers::failpoint_sleep_helper;
 /// * building in docker (either in CI or locally)
 ///
 /// One thing to note is that .git is not available in docker (and it is bad to include it there).
-/// When building locally, the `git_version` is used to query .git. When building on CI and docker,
-/// we don't build the actual PR branch commits, but always a "phantom" would be merge commit to
-/// the target branch -- the actual PR commit from which we build from is supplied as GIT_VERSION
-/// environment variable.
-///
-/// We ended up with this compromise between phantom would be merge commits vs. pull request branch
-/// heads due to old logs becoming more reliable (github could gc the phantom merge commit
-/// anytime) in #4641.
-///
-/// To avoid running buildscript every recompilation, we use rerun-if-env-changed option.
+/// So everything becides docker build is covered by git_version crate, and docker uses a `GIT_VERSION` argument to get the value required.
+/// It takes variable from build process env and puts it to the rustc env. And then we can retrieve it here by using env! macro.
+/// Git version received from environment variable used as a fallback in git_version invocation.
+/// And to avoid running buildscript every recompilation, we use rerun-if-env-changed option.
 /// So the build script will be run only when GIT_VERSION envvar has changed.
 ///
 /// Why not to use buildscript to get git commit sha directly without procmacro from different crate?
@@ -130,36 +124,25 @@ pub use failpoint_macro_helpers::failpoint_sleep_helper;
 /// Note that with git_version prefix is `git:` and in case of git version from env its `git-env:`.
 ///
 /// #############################################################################################
-/// TODO this macro is not the way the library is intended to be used, see <https://github.com/neondatabase/neon/issues/1565> for details.
-/// We use `cachepot` to reduce our current CI build times: <https://github.com/neondatabase/cloud/pull/1033#issuecomment-1100935036>
+/// TODO this macro is not the way the library is intended to be used, see https://github.com/neondatabase/neon/issues/1565 for details.
+/// We use `cachepot` to reduce our current CI build times: https://github.com/neondatabase/cloud/pull/1033#issuecomment-1100935036
 /// Yet, it seems to ignore the GIT_VERSION env variable, passed to Docker build, even with build.rs that contains
 /// `println!("cargo:rerun-if-env-changed=GIT_VERSION");` code for cachepot cache invalidation.
 /// The problem needs further investigation and regular `const` declaration instead of a macro.
 #[macro_export]
 macro_rules! project_git_version {
     ($const_identifier:ident) => {
-        // this should try GIT_VERSION first only then git_version::git_version!
-        const $const_identifier: &::core::primitive::str = {
-            const __COMMIT_FROM_GIT: &::core::primitive::str = git_version::git_version! {
-                prefix = "",
-                fallback = "unknown",
-                args = ["--abbrev=40", "--always", "--dirty=-modified"] // always use full sha
-            };
-
-            const __ARG: &[&::core::primitive::str; 2] = &match ::core::option_env!("GIT_VERSION") {
-                ::core::option::Option::Some(x) => ["git-env:", x],
-                ::core::option::Option::None => ["git:", __COMMIT_FROM_GIT],
-            };
-
-            $crate::__const_format::concatcp!(__ARG[0], __ARG[1])
-        };
+        const $const_identifier: &str = git_version::git_version!(
+            prefix = "git:",
+            fallback = concat!(
+                "git-env:",
+                env!("GIT_VERSION", "Missing GIT_VERSION envvar")
+            ),
+            args = ["--abbrev=40", "--always", "--dirty=-modified"] // always use full sha
+        );
     };
 }
 
-/// Re-export for `project_git_version` macro
-#[doc(hidden)]
-pub use const_format as __const_format;
-
 /// Same as `assert!`, but evaluated during compilation and gets optimized out in runtime.
 #[macro_export]
 macro_rules! const_assert {

libs/utils/src/lock_file.rs

@@ -1,10 +1,9 @@
 //! A module to create and read lock files.
 //!
 //! File locking is done using [`fcntl::flock`] exclusive locks.
-//! The only consumer of this module is currently
-//! [`pid_file`](crate::pid_file). See the module-level comment
-//! there for potential pitfalls with lock files that are used
-//! to store PIDs (pidfiles).
+//! The only consumer of this module is currently [`pid_file`].
+//! See the module-level comment there for potential pitfalls
+//! with lock files that are used to store PIDs (pidfiles).
 
 use std::{
     fs,
@@ -82,7 +81,7 @@ pub fn create_exclusive(lock_file_path: &Path) -> anyhow::Result<UnwrittenLockFi
 }
 
 /// Returned by [`read_and_hold_lock_file`].
-/// Check out the [`pid_file`](crate::pid_file) module for what the variants mean
+/// Check out the [`pid_file`] module for what the variants mean
 /// and potential caveats if the lock files that are used to store PIDs.
 pub enum LockFileRead {
     /// No file exists at the given path.

libs/utils/src/logging.rs

@@ -112,7 +112,7 @@ pub fn init(
 ///
 /// When the return value is dropped, the hook is reverted to std default hook (prints to stderr).
 /// If the assumptions about the initialization order are not held, use
-/// [`TracingPanicHookGuard::forget`] but keep in mind, if tracing is stopped, then panics will be
+/// [`TracingPanicHookGuard::disarm`] but keep in mind, if tracing is stopped, then panics will be
 /// lost.
 #[must_use]
 pub fn replace_panic_hook_with_tracing_panic_hook() -> TracingPanicHookGuard {

libs/utils/src/measured_stream.rs

@@ -1,5 +1,4 @@
 use pin_project_lite::pin_project;
-use std::io::Read;
 use std::pin::Pin;
 use std::{io, task};
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
@@ -76,34 +75,3 @@ impl<S: AsyncWrite + Unpin, R, W: FnMut(usize)> AsyncWrite for MeasuredStream<S,
         self.project().stream.poll_shutdown(context)
     }
 }
-
-/// Wrapper for a reader that counts bytes read.
-///
-/// Similar to MeasuredStream but it's one way and it's sync
-pub struct MeasuredReader<R: Read> {
-    inner: R,
-    byte_count: usize,
-}
-
-impl<R: Read> MeasuredReader<R> {
-    pub fn new(reader: R) -> Self {
-        Self {
-            inner: reader,
-            byte_count: 0,
-        }
-    }
-
-    pub fn get_byte_count(&self) -> usize {
-        self.byte_count
-    }
-}
-
-impl<R: Read> Read for MeasuredReader<R> {
-    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
-        let result = self.inner.read(buf);
-        if let Ok(n_bytes) = result {
-            self.byte_count += n_bytes
-        }
-        result
-    }
-}

libs/utils/src/seqwait.rs

@@ -23,9 +23,9 @@ pub enum SeqWaitError {
 
 /// Monotonically increasing value
 ///
-/// It is handy to store some other fields under the same mutex in `SeqWait<S>`
+/// It is handy to store some other fields under the same mutex in SeqWait<S>
 /// (e.g. store prev_record_lsn). So we allow SeqWait to be parametrized with
-/// any type that can expose counter. `V` is the type of exposed counter.
+/// any type that can expose counter. <V> is the type of exposed counter.
 pub trait MonotonicCounter<V> {
     /// Bump counter value and check that it goes forward
     /// N.B.: new_val is an actual new value, not a difference.
@@ -90,7 +90,7 @@ impl<T: Ord> Eq for Waiter<T> {}
 /// [`wait_for`]: SeqWait::wait_for
 /// [`advance`]: SeqWait::advance
 ///
-/// `S` means Storage, `V` is type of counter that this storage exposes.
+/// <S> means Storage, <V> is type of counter that this storage exposes.
 ///
 pub struct SeqWait<S, V>
 where

libs/utils/src/tracing_span_assert.rs

@@ -1,15 +1,8 @@
 //! Assert that the current [`tracing::Span`] has a given set of fields.
 //!
-//! Can only produce meaningful positive results when tracing has been configured as in example.
-//! Absence of `tracing_error::ErrorLayer` is not detected yet.
-//!
-//! `#[cfg(test)]` code will get a pass when using the `check_fields_present` macro in case tracing
-//! is completly unconfigured.
-//!
 //! # Usage
 //!
-//! ```rust
-//! # fn main() {
+//! ```
 //! use tracing_subscriber::prelude::*;
 //! let registry = tracing_subscriber::registry()
 //!    .with(tracing_error::ErrorLayer::default());
@@ -27,18 +20,23 @@
 //!
 //! use utils::tracing_span_assert::{check_fields_present, MultiNameExtractor};
 //! let extractor = MultiNameExtractor::new("TestExtractor", ["test", "test_id"]);
-//! if let Err(missing) = check_fields_present!([&extractor]) {
-//!    // if you copypaste this to a custom assert method, remember to add #[track_caller]
-//!    // to get the "user" code location for the panic.
-//!    panic!("Missing fields: {missing:?}");
+//! match check_fields_present([&extractor]) {
+//!    Ok(()) => {},
+//!    Err(missing) => {
+//!        panic!("Missing fields: {:?}", missing.into_iter().map(|f| f.name() ).collect::<Vec<_>>());
+//!    }
 //! }
-//! # }
 //! ```
 //!
-//! Recommended reading: <https://docs.rs/tracing-subscriber/0.3.16/tracing_subscriber/layer/index.html#per-layer-filtering>
+//! Recommended reading: https://docs.rs/tracing-subscriber/0.3.16/tracing_subscriber/layer/index.html#per-layer-filtering
 //!
 
-#[derive(Debug)]
+use std::{
+    collections::HashSet,
+    fmt::{self},
+    hash::{Hash, Hasher},
+};
+
 pub enum ExtractionResult {
     Present,
     Absent,
@@ -73,103 +71,49 @@ impl<const L: usize> Extractor for MultiNameExtractor<L> {
     }
 }
 
-/// Checks that the given extractors are satisfied with the current span hierarchy.
-///
-/// This should not be called directly, but used through [`check_fields_present`] which allows
-/// `Summary::Unconfigured` only when the calling crate is being `#[cfg(test)]` as a conservative default.
-#[doc(hidden)]
-pub fn check_fields_present0<const L: usize>(
-    must_be_present: [&dyn Extractor; L],
-) -> Result<Summary, Vec<&dyn Extractor>> {
-    let mut missing = must_be_present.into_iter().collect::<Vec<_>>();
-    let trace = tracing_error::SpanTrace::capture();
-    trace.with_spans(|md, _formatted_fields| {
-        // when trying to understand the inner workings of how does the matching work, note that
-        // this closure might be called zero times if the span is disabled. normally it is called
-        // once per span hierarchy level.
-        missing.retain(|extractor| match extractor.extract(md.fields()) {
-            ExtractionResult::Present => false,
-            ExtractionResult::Absent => true,
-        });
+struct MemoryIdentity<'a>(&'a dyn Extractor);
 
-        // continue walking up until we've found all missing
-        !missing.is_empty()
-    });
-    if missing.is_empty() {
-        Ok(Summary::FoundEverything)
-    } else if !tracing_subscriber_configured() {
-        Ok(Summary::Unconfigured)
-    } else {
-        // we can still hit here if a tracing subscriber has been configured but the ErrorLayer is
-        // missing, which can be annoying. for this case, we could probably use
-        // SpanTrace::status().
-        //
-        // another way to end up here is with RUST_LOG=pageserver=off while configuring the
-        // logging, though I guess in that case the SpanTrace::status() == EMPTY would be valid.
-        // this case is covered by test `not_found_if_tracing_error_subscriber_has_wrong_filter`.
-        Err(missing)
+impl<'a> MemoryIdentity<'a> {
+    fn as_ptr(&self) -> *const () {
+        self.0 as *const _ as *const ()
+    }
+}
+impl<'a> PartialEq for MemoryIdentity<'a> {
+    fn eq(&self, other: &Self) -> bool {
+        self.as_ptr() == other.as_ptr()
+    }
+}
+impl<'a> Eq for MemoryIdentity<'a> {}
+impl<'a> Hash for MemoryIdentity<'a> {
+    fn hash<H: Hasher>(&self, state: &mut H) {
+        self.as_ptr().hash(state);
+    }
+}
+impl<'a> fmt::Debug for MemoryIdentity<'a> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{:p}: {}", self.as_ptr(), self.0.name())
     }
 }
 
-/// Checks that the given extractors are satisfied with the current span hierarchy.
-///
-/// The macro is the preferred way of checking if fields exist while passing checks if a test does
-/// not have tracing configured.
-///
-/// Why mangled name? Because #[macro_export] will expose it at utils::__check_fields_present.
-/// However we can game a module namespaced macro for `use` purposes by re-exporting the
-/// #[macro_export] exported name with an alias (below).
-#[doc(hidden)]
-#[macro_export]
-macro_rules! __check_fields_present {
-    ($extractors:expr) => {{
-        {
-            use $crate::tracing_span_assert::{check_fields_present0, Summary::*, Extractor};
-
-            match check_fields_present0($extractors) {
-                Ok(FoundEverything) => Ok(()),
-                Ok(Unconfigured) if cfg!(test) => {
-                    // allow unconfigured in tests
-                    Ok(())
-                },
-                Ok(Unconfigured) => {
-                    panic!("utils::tracing_span_assert: outside of #[cfg(test)] expected tracing to be configured with tracing_error::ErrorLayer")
-                },
-                Err(missing) => Err(missing)
-            }
-        }
-    }}
-}
-
-pub use crate::__check_fields_present as check_fields_present;
-
-/// Explanation for why the check was deemed ok.
-///
-/// Mainly useful for testing, or configuring per-crate behaviour as in with
-/// [`check_fields_present`].
-#[derive(Debug)]
-pub enum Summary {
-    /// All extractors were found.
-    ///
-    /// Should only happen when tracing is properly configured.
-    FoundEverything,
-
-    /// Tracing has not been configured at all. This is ok for tests running without tracing set
-    /// up.
-    Unconfigured,
-}
-
-fn tracing_subscriber_configured() -> bool {
-    let mut noop_configured = false;
-    tracing::dispatcher::get_default(|d| {
-        // it is possible that this closure will not be invoked, but the current implementation
-        // always invokes it
-        noop_configured = d
-            .downcast_ref::<tracing::subscriber::NoSubscriber>()
-            .is_some();
+/// The extractor names passed as keys to [`new`].
+pub fn check_fields_present<const L: usize>(
+    must_be_present: [&dyn Extractor; L],
+) -> Result<(), Vec<&dyn Extractor>> {
+    let mut missing: HashSet<MemoryIdentity> =
+        HashSet::from_iter(must_be_present.into_iter().map(|r| MemoryIdentity(r)));
+    let trace = tracing_error::SpanTrace::capture();
+    trace.with_spans(|md, _formatted_fields| {
+        missing.retain(|extractor| match extractor.0.extract(md.fields()) {
+            ExtractionResult::Present => false,
+            ExtractionResult::Absent => true,
+        });
+        !missing.is_empty() // continue walking up until we've found all missing
     });
-
-    !noop_configured
+    if missing.is_empty() {
+        Ok(())
+    } else {
+        Err(missing.into_iter().map(|mi| mi.0).collect())
+    }
 }
 
 #[cfg(test)]
@@ -179,36 +123,6 @@ mod tests {
 
     use super::*;
 
-    use std::{
-        collections::HashSet,
-        fmt::{self},
-        hash::{Hash, Hasher},
-    };
-
-    struct MemoryIdentity<'a>(&'a dyn Extractor);
-
-    impl<'a> MemoryIdentity<'a> {
-        fn as_ptr(&self) -> *const () {
-            self.0 as *const _ as *const ()
-        }
-    }
-    impl<'a> PartialEq for MemoryIdentity<'a> {
-        fn eq(&self, other: &Self) -> bool {
-            self.as_ptr() == other.as_ptr()
-        }
-    }
-    impl<'a> Eq for MemoryIdentity<'a> {}
-    impl<'a> Hash for MemoryIdentity<'a> {
-        fn hash<H: Hasher>(&self, state: &mut H) {
-            self.as_ptr().hash(state);
-        }
-    }
-    impl<'a> fmt::Debug for MemoryIdentity<'a> {
-        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> std::fmt::Result {
-            write!(f, "{:p}: {}", self.as_ptr(), self.0.name())
-        }
-    }
-
     struct Setup {
         _current_thread_subscriber_guard: tracing::subscriber::DefaultGuard,
         tenant_extractor: MultiNameExtractor<2>,
@@ -245,8 +159,7 @@ mod tests {
         let setup = setup_current_thread();
         let span = tracing::info_span!("root", tenant_id = "tenant-1", timeline_id = "timeline-1");
         let _guard = span.enter();
-        let res = check_fields_present0([&setup.tenant_extractor, &setup.timeline_extractor]);
-        assert!(matches!(res, Ok(Summary::FoundEverything)), "{res:?}");
+        check_fields_present([&setup.tenant_extractor, &setup.timeline_extractor]).unwrap();
     }
 
     #[test]
@@ -254,8 +167,8 @@ mod tests {
         let setup = setup_current_thread();
         let span = tracing::info_span!("root", timeline_id = "timeline-1");
         let _guard = span.enter();
-        let missing = check_fields_present0([&setup.tenant_extractor, &setup.timeline_extractor])
-            .unwrap_err();
+        let missing =
+            check_fields_present([&setup.tenant_extractor, &setup.timeline_extractor]).unwrap_err();
         assert_missing(missing, vec![&setup.tenant_extractor]);
     }
 
@@ -272,8 +185,7 @@ mod tests {
         let span = tracing::info_span!("grandchild", timeline_id = "timeline-1");
         let _guard = span.enter();
 
-        let res = check_fields_present0([&setup.tenant_extractor, &setup.timeline_extractor]);
-        assert!(matches!(res, Ok(Summary::FoundEverything)), "{res:?}");
+        check_fields_present([&setup.tenant_extractor, &setup.timeline_extractor]).unwrap();
     }
 
     #[test]
@@ -286,7 +198,7 @@ mod tests {
         let span = tracing::info_span!("child", timeline_id = "timeline-1");
         let _guard = span.enter();
 
-        let missing = check_fields_present0([&setup.tenant_extractor]).unwrap_err();
+        let missing = check_fields_present([&setup.tenant_extractor]).unwrap_err();
         assert_missing(missing, vec![&setup.tenant_extractor]);
     }
 
@@ -295,8 +207,7 @@ mod tests {
         let setup = setup_current_thread();
         let span = tracing::info_span!("root", tenant_id = "tenant-1", timeline_id = "timeline-1");
         let _guard = span.enter();
-        let res = check_fields_present0([&setup.tenant_extractor]);
-        assert!(matches!(res, Ok(Summary::FoundEverything)), "{res:?}");
+        check_fields_present([&setup.tenant_extractor]).unwrap();
     }
 
     #[test]
@@ -312,8 +223,7 @@ mod tests {
         let span = tracing::info_span!("grandchild", timeline_id = "timeline-1");
         let _guard = span.enter();
 
-        let res = check_fields_present0([&setup.tenant_extractor]);
-        assert!(matches!(res, Ok(Summary::FoundEverything)), "{res:?}");
+        check_fields_present([&setup.tenant_extractor]).unwrap();
     }
 
     #[test]
@@ -321,7 +231,7 @@ mod tests {
         let setup = setup_current_thread();
         let span = tracing::info_span!("root", timeline_id = "timeline-1");
         let _guard = span.enter();
-        let missing = check_fields_present0([&setup.tenant_extractor]).unwrap_err();
+        let missing = check_fields_present([&setup.tenant_extractor]).unwrap_err();
         assert_missing(missing, vec![&setup.tenant_extractor]);
     }
 
@@ -335,107 +245,43 @@ mod tests {
         let span = tracing::info_span!("child", timeline_id = "timeline-1");
         let _guard = span.enter();
 
-        let missing = check_fields_present0([&setup.tenant_extractor]).unwrap_err();
+        let missing = check_fields_present([&setup.tenant_extractor]).unwrap_err();
         assert_missing(missing, vec![&setup.tenant_extractor]);
     }
 
     #[test]
-    fn tracing_error_subscriber_not_set_up_straight_line() {
+    fn tracing_error_subscriber_not_set_up() {
         // no setup
+
         let span = tracing::info_span!("foo", e = "some value");
         let _guard = span.enter();
 
         let extractor = MultiNameExtractor::new("E", ["e"]);
-        let res = check_fields_present0([&extractor]);
-        assert!(matches!(res, Ok(Summary::Unconfigured)), "{res:?}");
-
-        // similarly for a not found key
-        let extractor = MultiNameExtractor::new("F", ["foobar"]);
-        let res = check_fields_present0([&extractor]);
-        assert!(matches!(res, Ok(Summary::Unconfigured)), "{res:?}");
+        let missing = check_fields_present([&extractor]).unwrap_err();
+        assert_missing(missing, vec![&extractor]);
     }
 
     #[test]
-    fn tracing_error_subscriber_not_set_up_with_instrument() {
-        // no setup
-
-        // demo a case where span entering is used to establish a parent child connection, but
-        // when we re-enter the subspan SpanTrace::with_spans iterates over nothing.
-        let span = tracing::info_span!("foo", e = "some value");
-        let _guard = span.enter();
-
-        let subspan = tracing::info_span!("bar", f = "foobar");
-        drop(_guard);
-
-        // normally this would work, but without any tracing-subscriber configured, both
-        // check_field_present find nothing
-        let _guard = subspan.enter();
-        let extractors: [&dyn Extractor; 2] = [
-            &MultiNameExtractor::new("E", ["e"]),
-            &MultiNameExtractor::new("F", ["f"]),
-        ];
-
-        let res = check_fields_present0(extractors);
-        assert!(matches!(res, Ok(Summary::Unconfigured)), "{res:?}");
-
-        // similarly for a not found key
-        let extractor = MultiNameExtractor::new("G", ["g"]);
-        let res = check_fields_present0([&extractor]);
-        assert!(matches!(res, Ok(Summary::Unconfigured)), "{res:?}");
-    }
-
-    #[test]
-    fn tracing_subscriber_configured() {
-        // this will fail if any utils::logging::init callers appear, but let's hope they do not
-        // appear.
-        assert!(!super::tracing_subscriber_configured());
-
-        let _g = setup_current_thread();
-
-        assert!(super::tracing_subscriber_configured());
-    }
-
-    #[test]
-    fn not_found_when_disabled_by_filter() {
+    #[should_panic]
+    fn panics_if_tracing_error_subscriber_has_wrong_filter() {
         let r = tracing_subscriber::registry().with({
-            tracing_error::ErrorLayer::default().with_filter(tracing_subscriber::filter::filter_fn(
-                |md| !(md.is_span() && *md.level() == tracing::Level::INFO),
-            ))
+            tracing_error::ErrorLayer::default().with_filter(
+                tracing_subscriber::filter::dynamic_filter_fn(|md, _| {
+                    if md.is_span() && *md.level() == tracing::Level::INFO {
+                        return false;
+                    }
+                    true
+                }),
+            )
         });
 
         let _guard = tracing::subscriber::set_default(r);
 
-        // this test is a rather tricky one, it has a number of possible outcomes depending on the
-        // execution order when executed with other tests even if no test sets the global default
-        // subscriber.
-
         let span = tracing::info_span!("foo", e = "some value");
         let _guard = span.enter();
 
-        let extractors: [&dyn Extractor; 1] = [&MultiNameExtractor::new("E", ["e"])];
-
-        if span.is_disabled() {
-            // the tests are running single threaded, or we got lucky and no other tests subscriber
-            // was got to register their per-CALLSITE::META interest between `set_default` and
-            // creation of the span, thus the filter got to apply and registered interest of Never,
-            // so the span was never created.
-            //
-            // as the span is disabled, no keys were recorded to it, leading check_fields_present0
-            // to find an error.
-
-            let missing = check_fields_present0(extractors).unwrap_err();
-            assert_missing(missing, vec![extractors[0]]);
-        } else {
-            // when the span is enabled, it is because some other test is running at the same time,
-            // and that tests registry has filters which are interested in our above span.
-            //
-            // because the span is now enabled, all keys will be found for it. the
-            // tracing_error::SpanTrace does not consider layer filters during the span hierarchy
-            // walk (SpanTrace::with_spans), nor is the SpanTrace::status a reliable indicator in
-            // this test-induced issue.
-
-            let res = check_fields_present0(extractors);
-            assert!(matches!(res, Ok(Summary::FoundEverything)), "{res:?}");
-        }
+        let extractor = MultiNameExtractor::new("E", ["e"]);
+        let missing = check_fields_present([&extractor]).unwrap_err();
+        assert_missing(missing, vec![&extractor]);
     }
 }

pageserver/Cargo.toml

@@ -12,7 +12,6 @@ testing = ["fail/failpoints"]
 
 [dependencies]
 anyhow.workspace = true
-async-compression.workspace = true
 async-stream.workspace = true
 async-trait.workspace = true
 byteorder.workspace = true
@@ -25,7 +24,6 @@ consumption_metrics.workspace = true
 crc32c.workspace = true
 crossbeam-utils.workspace = true
 either.workspace = true
-flate2.workspace = true
 fail.workspace = true
 futures.workspace = true
 git-version.workspace = true

pageserver/benches/bench_layer_map.rs

@@ -1,8 +1,8 @@
 use pageserver::keyspace::{KeyPartitioning, KeySpace};
 use pageserver::repository::Key;
 use pageserver::tenant::layer_map::LayerMap;
-use pageserver::tenant::storage_layer::LayerFileName;
-use pageserver::tenant::storage_layer::PersistentLayerDesc;
+use pageserver::tenant::storage_layer::{tests::LayerDescriptor, Layer, LayerFileName};
+use pageserver::tenant::storage_layer::{PersistentLayer, PersistentLayerDesc};
 use rand::prelude::{SeedableRng, SliceRandom, StdRng};
 use std::cmp::{max, min};
 use std::fs::File;
@@ -28,13 +28,13 @@ fn build_layer_map(filename_dump: PathBuf) -> LayerMap {
     for fname in filenames {
         let fname = fname.unwrap();
         let fname = LayerFileName::from_str(&fname).unwrap();
-        let layer = PersistentLayerDesc::from(fname);
+        let layer = LayerDescriptor::from(fname);
 
         let lsn_range = layer.get_lsn_range();
         min_lsn = min(min_lsn, lsn_range.start);
         max_lsn = max(max_lsn, Lsn(lsn_range.end.0 - 1));
 
-        updates.insert_historic(layer);
+        updates.insert_historic(layer.layer_desc().clone());
     }
 
     println!("min: {min_lsn}, max: {max_lsn}");
@@ -210,15 +210,15 @@ fn bench_sequential(c: &mut Criterion) {
     for i in 0..100_000 {
         let i32 = (i as u32) % 100;
         let zero = Key::from_hex("000000000000000000000000000000000000").unwrap();
-        let layer = PersistentLayerDesc::new_img(
+        let layer = LayerDescriptor::from(PersistentLayerDesc::new_img(
             TenantId::generate(),
             TimelineId::generate(),
             zero.add(10 * i32)..zero.add(10 * i32 + 1),
             Lsn(i),
             false,
             0,
-        );
-        updates.insert_historic(layer);
+        ));
+        updates.insert_historic(layer.layer_desc().clone());
     }
     updates.flush();
     println!("Finished layer map init in {:?}", now.elapsed());

pageserver/ctl/src/draw_timeline_dir.rs

@@ -7,10 +7,10 @@
 //! - The y axis represents LSN, growing upwards.
 //!
 //! Coordinates in both axis are compressed for better readability.
-//! (see <https://medium.com/algorithms-digest/coordinate-compression-2fff95326fb>)
+//! (see https://medium.com/algorithms-digest/coordinate-compression-2fff95326fb)
 //!
 //! Example use:
-//! ```bash
+//! ```
 //! $ ls test_output/test_pgbench\[neon-45-684\]/repo/tenants/$TENANT/timelines/$TIMELINE | \
 //! $   grep "__" | cargo run --release --bin pagectl draw-timeline-dir > out.svg
 //! $ firefox out.svg
@@ -20,7 +20,7 @@
 //! or from pageserver log files.
 //!
 //! TODO Consider shipping this as a grafana panel plugin:
-//!      <https://grafana.com/tutorials/build-a-panel-plugin/>
+//!      https://grafana.com/tutorials/build-a-panel-plugin/
 use anyhow::Result;
 use pageserver::repository::Key;
 use std::cmp::Ordering;
@@ -117,8 +117,7 @@ pub fn main() -> Result<()> {
 
         let mut lsn_diff = (lsn_end - lsn_start) as f32;
         let mut fill = Fill::None;
-        let mut ymargin = 0.05 * lsn_diff; // Height-dependent margin to disambiguate overlapping deltas
-        let xmargin = 0.05; // Height-dependent margin to disambiguate overlapping deltas
+        let mut margin = 0.05 * lsn_diff; // Height-dependent margin to disambiguate overlapping deltas
         let mut lsn_offset = 0.0;
 
         // Fill in and thicken rectangle if it's an
@@ -129,7 +128,7 @@ pub fn main() -> Result<()> {
                 num_images += 1;
                 lsn_diff = 0.3;
                 lsn_offset = -lsn_diff / 2.0;
-                ymargin = 0.05;
+                margin = 0.05;
                 fill = Fill::Color(rgb(0, 0, 0));
             }
             Ordering::Greater => panic!("Invalid lsn range {}-{}", lsn_start, lsn_end),
@@ -138,10 +137,10 @@ pub fn main() -> Result<()> {
         println!(
             "    {}",
             rectangle(
-                key_start as f32 + stretch * xmargin,
-                stretch * (lsn_max as f32 - (lsn_end as f32 - ymargin - lsn_offset)),
-                key_diff as f32 - stretch * 2.0 * xmargin,
-                stretch * (lsn_diff - 2.0 * ymargin)
+                key_start as f32 + stretch * margin,
+                stretch * (lsn_max as f32 - (lsn_end as f32 - margin - lsn_offset)),
+                key_diff as f32 - stretch * 2.0 * margin,
+                stretch * (lsn_diff - 2.0 * margin)
             )
             .fill(fill)
             .stroke(Stroke::Color(rgb(0, 0, 0), 0.1))

pageserver/src/config.rs

@@ -171,13 +171,11 @@ pub struct PageServerConf {
 
     pub log_format: LogFormat,
 
-    /// Number of concurrent [`Tenant::gather_size_inputs`](crate::tenant::Tenant::gather_size_inputs) allowed.
+    /// Number of concurrent [`Tenant::gather_size_inputs`] allowed.
     pub concurrent_tenant_size_logical_size_queries: ConfigurableSemaphore,
     /// Limit of concurrent [`Tenant::gather_size_inputs`] issued by module `eviction_task`.
     /// The number of permits is the same as `concurrent_tenant_size_logical_size_queries`.
     /// See the comment in `eviction_task` for details.
-    ///
-    /// [`Tenant::gather_size_inputs`]: crate::tenant::Tenant::gather_size_inputs
     pub eviction_task_immitated_concurrent_logical_size_queries: ConfigurableSemaphore,
 
     // How often to collect metrics and send them to the metrics endpoint.
@@ -572,21 +570,21 @@ impl PageServerConf {
             .join(TENANT_ATTACHING_MARKER_FILENAME)
     }
 
-    pub fn tenant_ignore_mark_file_path(&self, tenant_id: &TenantId) -> PathBuf {
-        self.tenant_path(tenant_id).join(IGNORED_TENANT_FILE_NAME)
+    pub fn tenant_ignore_mark_file_path(&self, tenant_id: TenantId) -> PathBuf {
+        self.tenant_path(&tenant_id).join(IGNORED_TENANT_FILE_NAME)
     }
 
     /// Points to a place in pageserver's local directory,
     /// where certain tenant's tenantconf file should be located.
-    pub fn tenant_config_path(&self, tenant_id: &TenantId) -> PathBuf {
-        self.tenant_path(tenant_id).join(TENANT_CONFIG_NAME)
+    pub fn tenant_config_path(&self, tenant_id: TenantId) -> PathBuf {
+        self.tenant_path(&tenant_id).join(TENANT_CONFIG_NAME)
     }
 
     pub fn timelines_path(&self, tenant_id: &TenantId) -> PathBuf {
         self.tenant_path(tenant_id).join(TIMELINES_SEGMENT_NAME)
     }
 
-    pub fn timeline_path(&self, tenant_id: &TenantId, timeline_id: &TimelineId) -> PathBuf {
+    pub fn timeline_path(&self, timeline_id: &TimelineId, tenant_id: &TenantId) -> PathBuf {
         self.timelines_path(tenant_id).join(timeline_id.to_string())
     }
 
@@ -596,7 +594,7 @@ impl PageServerConf {
         timeline_id: TimelineId,
     ) -> PathBuf {
         path_with_suffix_extension(
-            self.timeline_path(&tenant_id, &timeline_id),
+            self.timeline_path(&timeline_id, &tenant_id),
             TIMELINE_UNINIT_MARK_SUFFIX,
         )
     }
@@ -619,8 +617,8 @@ impl PageServerConf {
 
     /// Points to a place in pageserver's local directory,
     /// where certain timeline's metadata file should be located.
-    pub fn metadata_path(&self, tenant_id: &TenantId, timeline_id: &TimelineId) -> PathBuf {
-        self.timeline_path(tenant_id, timeline_id)
+    pub fn metadata_path(&self, timeline_id: TimelineId, tenant_id: TenantId) -> PathBuf {
+        self.timeline_path(&timeline_id, &tenant_id)
             .join(METADATA_FILE_NAME)
     }
 
@@ -655,7 +653,6 @@ impl PageServerConf {
         match pg_version {
             14 => Ok(path.join(format!("v{pg_version}"))),
             15 => Ok(path.join(format!("v{pg_version}"))),
-            16 => Ok(path.join(format!("v{pg_version}"))),
             _ => bail!("Unsupported postgres version: {}", pg_version),
         }
     }
@@ -664,7 +661,6 @@ impl PageServerConf {
         match pg_version {
             14 => Ok(self.pg_distrib_dir(pg_version)?.join("bin")),
             15 => Ok(self.pg_distrib_dir(pg_version)?.join("bin")),
-            16 => Ok(self.pg_distrib_dir(pg_version)?.join("bin")),
             _ => bail!("Unsupported postgres version: {}", pg_version),
         }
     }
@@ -672,7 +668,6 @@ impl PageServerConf {
         match pg_version {
             14 => Ok(self.pg_distrib_dir(pg_version)?.join("lib")),
             15 => Ok(self.pg_distrib_dir(pg_version)?.join("lib")),
-            16 => Ok(self.pg_distrib_dir(pg_version)?.join("lib")),
             _ => bail!("Unsupported postgres version: {}", pg_version),
         }
     }
@@ -998,8 +993,6 @@ impl ConfigurableSemaphore {
     /// Require a non-zero initial permits, because using permits == 0 is a crude way to disable a
     /// feature such as [`Tenant::gather_size_inputs`]. Otherwise any semaphore using future will
     /// behave like [`futures::future::pending`], just waiting until new permits are added.
-    ///
-    /// [`Tenant::gather_size_inputs`]: crate::tenant::Tenant::gather_size_inputs
     pub fn new(initial_permits: NonZeroUsize) -> Self {
         ConfigurableSemaphore {
             initial_permits,

pageserver/src/consumption_metrics.rs

@@ -234,18 +234,14 @@ pub async fn collect_metrics_iteration(
         // Note that this metric is calculated in a separate bgworker
         // Here we only use cached value, which may lag behind the real latest one
         let tenant_synthetic_size = tenant.get_cached_synthetic_size();
-
-        if tenant_synthetic_size != 0 {
-            // only send non-zeroes because otherwise these show up as errors in logs
-            current_metrics.push((
-                PageserverConsumptionMetricsKey {
-                    tenant_id,
-                    timeline_id: None,
-                    metric: SYNTHETIC_STORAGE_SIZE,
-                },
-                tenant_synthetic_size,
-            ));
-        }
+        current_metrics.push((
+            PageserverConsumptionMetricsKey {
+                tenant_id,
+                timeline_id: None,
+                metric: SYNTHETIC_STORAGE_SIZE,
+            },
+            tenant_synthetic_size,
+        ));
     }
 
     // Filter metrics, unless we want to send all metrics, including cached ones.

pageserver/src/context.rs

@@ -179,9 +179,6 @@ impl RequestContext {
     /// a context and you are unwilling to change all callers to provide one.
     ///
     /// Before we add cancellation, we should get rid of this method.
-    ///
-    /// [`attached_child`]: Self::attached_child
-    /// [`detached_child`]: Self::detached_child
     pub fn todo_child(task_kind: TaskKind, download_behavior: DownloadBehavior) -> Self {
         Self::new(task_kind, download_behavior)
     }

pageserver/src/disk_usage_eviction_task.rs

@@ -305,7 +305,7 @@ pub async fn disk_usage_eviction_task_iteration_impl<U: Usage>(
     let now = SystemTime::now();
     for (i, (partition, candidate)) in candidates.iter().enumerate() {
         debug!(
-            "cand {}/{}: size={}, no_access_for={}us, partition={:?}, {}/{}/{}",
+            "cand {}/{}: size={}, no_access_for={}us, parition={:?}, tenant={} timeline={} layer={}",
             i + 1,
             candidates.len(),
             candidate.layer.file_size(),

pageserver/src/http/routes.rs

@@ -346,7 +346,7 @@ async fn timeline_create_handler(
             Err(tenant::CreateTimelineError::Other(err)) => Err(ApiError::InternalServerError(err)),
         }
     }
-    .instrument(info_span!("timeline_create", %tenant_id, timeline_id = %new_timeline_id, lsn=?request_data.ancestor_start_lsn, pg_version=?request_data.pg_version))
+    .instrument(info_span!("timeline_create", tenant = %tenant_id, timeline_id = %new_timeline_id, lsn=?request_data.ancestor_start_lsn, pg_version=?request_data.pg_version))
     .await
 }
 
@@ -381,7 +381,7 @@ async fn timeline_list_handler(
         }
         Ok::<Vec<TimelineInfo>, ApiError>(response_data)
     }
-    .instrument(info_span!("timeline_list", %tenant_id))
+    .instrument(info_span!("timeline_list", tenant = %tenant_id))
     .await?;
 
     json_response(StatusCode::OK, response_data)
@@ -418,7 +418,7 @@ async fn timeline_detail_handler(
 
         Ok::<_, ApiError>(timeline_info)
     }
-    .instrument(info_span!("timeline_detail", %tenant_id, %timeline_id))
+    .instrument(info_span!("timeline_detail", tenant = %tenant_id, timeline = %timeline_id))
     .await?;
 
     json_response(StatusCode::OK, timeline_info)
@@ -479,7 +479,7 @@ async fn tenant_attach_handler(
             remote_storage.clone(),
             &ctx,
         )
-        .instrument(info_span!("tenant_attach", %tenant_id))
+        .instrument(info_span!("tenant_attach", tenant = %tenant_id))
         .await?;
     } else {
         return Err(ApiError::BadRequest(anyhow!(
@@ -501,7 +501,7 @@ async fn timeline_delete_handler(
     let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
 
     mgr::delete_timeline(tenant_id, timeline_id, &ctx)
-        .instrument(info_span!("timeline_delete", %tenant_id, %timeline_id))
+        .instrument(info_span!("timeline_delete", tenant = %tenant_id, timeline = %timeline_id))
         .await?;
 
     // FIXME: needs to be an error for console to retry it. Ideally Accepted should be used and retried until 404.
@@ -519,7 +519,7 @@ async fn tenant_detach_handler(
     let state = get_state(&request);
     let conf = state.conf;
     mgr::detach_tenant(conf, tenant_id, detach_ignored.unwrap_or(false))
-        .instrument(info_span!("tenant_detach", %tenant_id))
+        .instrument(info_span!("tenant_detach", tenant = %tenant_id))
         .await?;
 
     json_response(StatusCode::OK, ())
@@ -542,7 +542,7 @@ async fn tenant_load_handler(
         state.remote_storage.clone(),
         &ctx,
     )
-    .instrument(info_span!("load", %tenant_id))
+    .instrument(info_span!("load", tenant = %tenant_id))
     .await?;
 
     json_response(StatusCode::ACCEPTED, ())
@@ -558,7 +558,7 @@ async fn tenant_ignore_handler(
     let state = get_state(&request);
     let conf = state.conf;
     mgr::ignore_tenant(conf, tenant_id)
-        .instrument(info_span!("ignore_tenant", %tenant_id))
+        .instrument(info_span!("ignore_tenant", tenant = %tenant_id))
         .await?;
 
     json_response(StatusCode::OK, ())
@@ -611,7 +611,7 @@ async fn tenant_status(
             attachment_status: state.attachment_status(),
         })
     }
-    .instrument(info_span!("tenant_status_handler", %tenant_id))
+    .instrument(info_span!("tenant_status_handler", tenant = %tenant_id))
     .await?;
 
     json_response(StatusCode::OK, tenant_info)
@@ -850,7 +850,7 @@ async fn tenant_create_handler(
         state.remote_storage.clone(),
         &ctx,
     )
-    .instrument(info_span!("tenant_create", tenant_id = %target_tenant_id))
+    .instrument(info_span!("tenant_create", tenant = ?target_tenant_id))
     .await?;
 
     // We created the tenant. Existing API semantics are that the tenant
@@ -912,7 +912,7 @@ async fn update_tenant_config_handler(
 
     let state = get_state(&request);
     mgr::set_new_tenant_config(state.conf, tenant_conf, tenant_id)
-        .instrument(info_span!("tenant_config", %tenant_id))
+        .instrument(info_span!("tenant_config", tenant = ?tenant_id))
         .await?;
 
     json_response(StatusCode::OK, ())
@@ -1143,7 +1143,7 @@ async fn disk_usage_eviction_run(
     let Some(storage) = state.remote_storage.clone() else {
         return Err(ApiError::InternalServerError(anyhow::anyhow!(
             "remote storage not configured, cannot run eviction iteration"
-        )));
+        )))
     };
 
     let state = state.disk_usage_eviction_state.clone();

pageserver/src/metrics.rs

@@ -385,7 +385,7 @@ pub static UNEXPECTED_ONDEMAND_DOWNLOADS: Lazy<IntCounter> = Lazy::new(|| {
     .expect("failed to define a metric")
 });
 
-/// Each `Timeline`'s  [`EVICTIONS_WITH_LOW_RESIDENCE_DURATION`] metric.
+/// Each [`Timeline`]'s  [`EVICTIONS_WITH_LOW_RESIDENCE_DURATION`] metric.
 #[derive(Debug)]
 pub struct EvictionsWithLowResidenceDuration {
     data_source: &'static str,
@@ -541,17 +541,6 @@ pub static SMGR_QUERY_TIME: Lazy<HistogramVec> = Lazy::new(|| {
     .expect("failed to define a metric")
 });
 
-// keep in sync with control plane Go code so that we can validate
-// compute's basebackup_ms metric with our perspective in the context of SLI/SLO.
-static COMPUTE_STARTUP_BUCKETS: Lazy<[f64; 28]> = Lazy::new(|| {
-    // Go code uses milliseconds. Variable is called `computeStartupBuckets`
-    [
-        5, 10, 20, 30, 50, 70, 100, 120, 150, 200, 250, 300, 350, 400, 450, 500, 600, 800, 1000,
-        1500, 2000, 2500, 3000, 5000, 10000, 20000, 40000, 60000,
-    ]
-    .map(|ms| (ms as f64) / 1000.0)
-});
-
 pub struct BasebackupQueryTime(HistogramVec);
 pub static BASEBACKUP_QUERY_TIME: Lazy<BasebackupQueryTime> = Lazy::new(|| {
     BasebackupQueryTime({
@@ -559,7 +548,7 @@ pub static BASEBACKUP_QUERY_TIME: Lazy<BasebackupQueryTime> = Lazy::new(|| {
             "pageserver_basebackup_query_seconds",
             "Histogram of basebackup queries durations, by result type",
             &["result"],
-            COMPUTE_STARTUP_BUCKETS.to_vec(),
+            CRITICAL_OP_BUCKETS.into(),
         )
         .expect("failed to define a metric")
     })
@@ -829,7 +818,7 @@ pub static WAL_REDO_RECORD_COUNTER: Lazy<IntCounter> = Lazy::new(|| {
     .unwrap()
 });
 
-/// Similar to `prometheus::HistogramTimer` but does not record on drop.
+/// Similar to [`prometheus::HistogramTimer`] but does not record on drop.
 pub struct StorageTimeMetricsTimer {
     metrics: StorageTimeMetrics,
     start: Instant,
@@ -887,7 +876,7 @@ impl StorageTimeMetrics {
 
     /// Starts timing a new operation.
     ///
-    /// Note: unlike `prometheus::HistogramTimer` the returned timer does not record on drop.
+    /// Note: unlike [`prometheus::HistogramTimer`] the returned timer does not record on drop.
     pub fn start_timer(&self) -> StorageTimeMetricsTimer {
         StorageTimeMetricsTimer::new(self.clone())
     }
@@ -1267,7 +1256,7 @@ impl RemoteTimelineClientMetrics {
     /// Update the metrics that change when a call to the remote timeline client instance starts.
     ///
     /// Drop the returned guard object once the operation is finished to updates corresponding metrics that track completions.
-    /// Or, use [`RemoteTimelineClientCallMetricGuard::will_decrement_manually`] and [`call_end`](Self::call_end) if that
+    /// Or, use [`RemoteTimelineClientCallMetricGuard::will_decrement_manually`] and [`call_end`] if that
     /// is more suitable.
     /// Never do both.
     pub(crate) fn call_begin(
@@ -1300,7 +1289,7 @@ impl RemoteTimelineClientMetrics {
 
     /// Manually udpate the metrics that track completions, instead of using the guard object.
     /// Using the guard object is generally preferable.
-    /// See [`call_begin`](Self::call_begin) for more context.
+    /// See [`call_begin`] for more context.
     pub(crate) fn call_end(
         &self,
         file_kind: &RemoteOpFileKind,

pageserver/src/page_service.rs

@@ -10,7 +10,6 @@
 //
 
 use anyhow::Context;
-use async_compression::tokio::write::GzipEncoder;
 use bytes::Buf;
 use bytes::Bytes;
 use futures::Stream;
@@ -32,10 +31,8 @@ use std::str;
 use std::str::FromStr;
 use std::sync::Arc;
 use std::time::Duration;
-use tokio::io::AsyncWriteExt;
 use tokio::io::{AsyncRead, AsyncWrite};
 use tokio_util::io::StreamReader;
-use tracing::field;
 use tracing::*;
 use utils::id::ConnectionId;
 use utils::{
@@ -54,7 +51,6 @@ use crate::metrics::{LIVE_CONNECTIONS_COUNT, SMGR_QUERY_TIME};
 use crate::task_mgr;
 use crate::task_mgr::TaskKind;
 use crate::tenant;
-use crate::tenant::debug_assert_current_span_has_tenant_and_timeline_id;
 use crate::tenant::mgr;
 use crate::tenant::mgr::GetTenantError;
 use crate::tenant::{Tenant, Timeline};
@@ -242,7 +238,6 @@ pub async fn libpq_listener_main(
     Ok(())
 }
 
-#[instrument(skip_all, fields(peer_addr))]
 async fn page_service_conn_main(
     conf: &'static PageServerConf,
     broker_client: storage_broker::BrokerClientChannel,
@@ -265,7 +260,6 @@ async fn page_service_conn_main(
         .context("could not set TCP_NODELAY")?;
 
     let peer_addr = socket.peer_addr().context("get peer address")?;
-    tracing::Span::current().record("peer_addr", field::display(peer_addr));
 
     // setup read timeout of 10 minutes. the timeout is rather arbitrary for requirements:
     // - long enough for most valid compute connections
@@ -368,7 +362,7 @@ impl PageServerHandler {
         }
     }
 
-    #[instrument(skip_all)]
+    #[instrument(skip(self, pgb, ctx))]
     async fn handle_pagerequests<IO>(
         &self,
         pgb: &mut PostgresBackend<IO>,
@@ -379,8 +373,6 @@ impl PageServerHandler {
     where
         IO: AsyncRead + AsyncWrite + Send + Sync + Unpin,
     {
-        debug_assert_current_span_has_tenant_and_timeline_id();
-
         // NOTE: pagerequests handler exits when connection is closed,
         //       so there is no need to reset the association
         task_mgr::associate_with(Some(tenant_id), Some(timeline_id));
@@ -481,7 +473,7 @@ impl PageServerHandler {
     }
 
     #[allow(clippy::too_many_arguments)]
-    #[instrument(skip_all, fields(%base_lsn, end_lsn=%_end_lsn, %pg_version))]
+    #[instrument(skip(self, pgb, ctx))]
     async fn handle_import_basebackup<IO>(
         &self,
         pgb: &mut PostgresBackend<IO>,
@@ -495,8 +487,6 @@ impl PageServerHandler {
     where
         IO: AsyncRead + AsyncWrite + Send + Sync + Unpin,
     {
-        debug_assert_current_span_has_tenant_and_timeline_id();
-
         task_mgr::associate_with(Some(tenant_id), Some(timeline_id));
         // Create empty timeline
         info!("creating new timeline");
@@ -541,7 +531,7 @@ impl PageServerHandler {
         Ok(())
     }
 
-    #[instrument(skip_all, fields(%start_lsn, %end_lsn))]
+    #[instrument(skip(self, pgb, ctx))]
     async fn handle_import_wal<IO>(
         &self,
         pgb: &mut PostgresBackend<IO>,
@@ -554,7 +544,6 @@ impl PageServerHandler {
     where
         IO: AsyncRead + AsyncWrite + Send + Sync + Unpin,
     {
-        debug_assert_current_span_has_tenant_and_timeline_id();
         task_mgr::associate_with(Some(tenant_id), Some(timeline_id));
 
         let timeline = get_active_tenant_timeline(tenant_id, timeline_id, &ctx).await?;
@@ -749,7 +738,7 @@ impl PageServerHandler {
     }
 
     #[allow(clippy::too_many_arguments)]
-    #[instrument(skip_all, fields(?lsn, ?prev_lsn, %full_backup))]
+    #[instrument(skip(self, pgb, ctx))]
     async fn handle_basebackup_request<IO>(
         &mut self,
         pgb: &mut PostgresBackend<IO>,
@@ -758,14 +747,11 @@ impl PageServerHandler {
         lsn: Option<Lsn>,
         prev_lsn: Option<Lsn>,
         full_backup: bool,
-        gzip: bool,
         ctx: RequestContext,
     ) -> anyhow::Result<()>
     where
         IO: AsyncRead + AsyncWrite + Send + Sync + Unpin,
     {
-        debug_assert_current_span_has_tenant_and_timeline_id();
-
         let started = std::time::Instant::now();
 
         // check that the timeline exists
@@ -786,9 +772,8 @@ impl PageServerHandler {
         pgb.write_message_noflush(&BeMessage::CopyOutResponse)?;
         pgb.flush().await?;
 
-        // Send a tarball of the latest layer on the timeline. Compress if not
-        // fullbackup. TODO Compress in that case too (tests need to be updated)
-        if full_backup {
+        // Send a tarball of the latest layer on the timeline
+        {
             let mut writer = pgb.copyout_writer();
             basebackup::send_basebackup_tarball(
                 &mut writer,
@@ -799,40 +784,6 @@ impl PageServerHandler {
                 &ctx,
             )
             .await?;
-        } else {
-            let mut writer = pgb.copyout_writer();
-            if gzip {
-                let mut encoder = GzipEncoder::with_quality(
-                    writer,
-                    // NOTE using fast compression because it's on the critical path
-                    //      for compute startup. For an empty database, we get
-                    //      <100KB with this method. The Level::Best compression method
-                    //      gives us <20KB, but maybe we should add basebackup caching
-                    //      on compute shutdown first.
-                    async_compression::Level::Fastest,
-                );
-                basebackup::send_basebackup_tarball(
-                    &mut encoder,
-                    &timeline,
-                    lsn,
-                    prev_lsn,
-                    full_backup,
-                    &ctx,
-                )
-                .await?;
-                // shutdown the encoder to ensure the gzip footer is written
-                encoder.shutdown().await?;
-            } else {
-                basebackup::send_basebackup_tarball(
-                    &mut writer,
-                    &timeline,
-                    lsn,
-                    prev_lsn,
-                    full_backup,
-                    &ctx,
-                )
-                .await?;
-            }
         }
 
         pgb.write_message_noflush(&BeMessage::CopyDone)?;
@@ -911,7 +862,6 @@ where
         Ok(())
     }
 
-    #[instrument(skip_all, fields(tenant_id, timeline_id))]
     async fn process_query(
         &mut self,
         pgb: &mut PostgresBackend<IO>,
@@ -933,10 +883,6 @@ where
             let timeline_id = TimelineId::from_str(params[1])
                 .with_context(|| format!("Failed to parse timeline id from {}", params[1]))?;
 
-            tracing::Span::current()
-                .record("tenant_id", field::display(tenant_id))
-                .record("timeline_id", field::display(timeline_id));
-
             self.check_permission(Some(tenant_id))?;
 
             self.handle_pagerequests(pgb, tenant_id, timeline_id, ctx)
@@ -956,10 +902,6 @@ where
             let timeline_id = TimelineId::from_str(params[1])
                 .with_context(|| format!("Failed to parse timeline id from {}", params[1]))?;
 
-            tracing::Span::current()
-                .record("tenant_id", field::display(tenant_id))
-                .record("timeline_id", field::display(timeline_id));
-
             self.check_permission(Some(tenant_id))?;
 
             let lsn = if params.len() >= 3 {
@@ -971,19 +913,6 @@ where
                 None
             };
 
-            let gzip = if params.len() >= 4 {
-                if params[3] == "--gzip" {
-                    true
-                } else {
-                    return Err(QueryError::Other(anyhow::anyhow!(
-                        "Parameter in position 3 unknown {}",
-                        params[3],
-                    )));
-                }
-            } else {
-                false
-            };
-
             metrics::metric_vec_duration::observe_async_block_duration_by_result(
                 &*crate::metrics::BASEBACKUP_QUERY_TIME,
                 async move {
@@ -994,7 +923,6 @@ where
                         lsn,
                         None,
                         false,
-                        gzip,
                         ctx,
                     )
                     .await?;
@@ -1020,10 +948,6 @@ where
             let timeline_id = TimelineId::from_str(params[1])
                 .with_context(|| format!("Failed to parse timeline id from {}", params[1]))?;
 
-            tracing::Span::current()
-                .record("tenant_id", field::display(tenant_id))
-                .record("timeline_id", field::display(timeline_id));
-
             self.check_permission(Some(tenant_id))?;
             let timeline = get_active_tenant_timeline(tenant_id, timeline_id, &ctx).await?;
 
@@ -1055,10 +979,6 @@ where
             let timeline_id = TimelineId::from_str(params[1])
                 .with_context(|| format!("Failed to parse timeline id from {}", params[1]))?;
 
-            tracing::Span::current()
-                .record("tenant_id", field::display(tenant_id))
-                .record("timeline_id", field::display(timeline_id));
-
             // The caller is responsible for providing correct lsn and prev_lsn.
             let lsn = if params.len() > 2 {
                 Some(
@@ -1080,17 +1000,8 @@ where
             self.check_permission(Some(tenant_id))?;
 
             // Check that the timeline exists
-            self.handle_basebackup_request(
-                pgb,
-                tenant_id,
-                timeline_id,
-                lsn,
-                prev_lsn,
-                true,
-                false,
-                ctx,
-            )
-            .await?;
+            self.handle_basebackup_request(pgb, tenant_id, timeline_id, lsn, prev_lsn, true, ctx)
+                .await?;
             pgb.write_message_noflush(&BeMessage::CommandComplete(b"SELECT 1"))?;
         } else if query_string.starts_with("import basebackup ") {
             // Import the `base` section (everything but the wal) of a basebackup.
@@ -1122,10 +1033,6 @@ where
             let pg_version = u32::from_str(params[4])
                 .with_context(|| format!("Failed to parse pg_version from {}", params[4]))?;
 
-            tracing::Span::current()
-                .record("tenant_id", field::display(tenant_id))
-                .record("timeline_id", field::display(timeline_id));
-
             self.check_permission(Some(tenant_id))?;
 
             match self
@@ -1170,10 +1077,6 @@ where
             let end_lsn = Lsn::from_str(params[3])
                 .with_context(|| format!("Failed to parse Lsn from {}", params[3]))?;
 
-            tracing::Span::current()
-                .record("tenant_id", field::display(tenant_id))
-                .record("timeline_id", field::display(timeline_id));
-
             self.check_permission(Some(tenant_id))?;
 
             match self
@@ -1205,8 +1108,6 @@ where
             let tenant_id = TenantId::from_str(params[0])
                 .with_context(|| format!("Failed to parse tenant id from {}", params[0]))?;
 
-            tracing::Span::current().record("tenant_id", field::display(tenant_id));
-
             self.check_permission(Some(tenant_id))?;
 
             let tenant = get_active_tenant_with_timeout(tenant_id, &ctx).await?;

pageserver/src/pgdatadir_mapping.rs

@@ -1131,7 +1131,7 @@ impl<'a> DatadirModification<'a> {
     /// context, breaking the atomicity is OK. If the import is interrupted, the
     /// whole import fails and the timeline will be deleted anyway.
     /// (Or to be precise, it will be left behind for debugging purposes and
-    /// ignored, see <https://github.com/neondatabase/neon/pull/1809>)
+    /// ignored, see https://github.com/neondatabase/neon/pull/1809)
     ///
     /// Note: A consequence of flushing the pending operations is that they
     /// won't be visible to subsequent operations until `commit`. The function

pageserver/src/task_mgr.rs

@@ -205,7 +205,7 @@ pub enum TaskKind {
     ///
     /// Walreceiver uses its own abstraction called `TaskHandle` to represent the activity of establishing and handling a connection.
     /// That abstraction doesn't use `task_mgr`.
-    /// The `WalReceiverManager` task ensures that this `TaskHandle` task does not outlive the `WalReceiverManager` task.
+    /// The [`WalReceiverManager`] task ensures that this `TaskHandle` task does not outlive the [`WalReceiverManager`] task.
     /// For the `RequestContext` that we hand to the TaskHandle, we use the [`WalReceiverConnectionHandler`] task kind.
     ///
     /// Once the connection is established, the `TaskHandle` task creates a
@@ -213,21 +213,16 @@ pub enum TaskKind {
     /// the `Connection` object.
     /// A `CancellationToken` created by the `TaskHandle` task ensures
     /// that the [`WalReceiverConnectionPoller`] task will cancel soon after as the `TaskHandle` is dropped.
-    ///
-    /// [`WalReceiverConnectionHandler`]: Self::WalReceiverConnectionHandler
-    /// [`WalReceiverConnectionPoller`]: Self::WalReceiverConnectionPoller
     WalReceiverManager,
 
-    /// The `TaskHandle` task that executes `handle_walreceiver_connection`.
+    /// The `TaskHandle` task that executes [`walreceiver_connection::handle_walreceiver_connection`].
     /// Not a `task_mgr` task, but we use this `TaskKind` for its `RequestContext`.
     /// See the comment on [`WalReceiverManager`].
-    ///
-    /// [`WalReceiverManager`]: Self::WalReceiverManager
     WalReceiverConnectionHandler,
 
     /// The task that polls the `tokio-postgres::Connection` object.
-    /// Spawned by task [`WalReceiverConnectionHandler`](Self::WalReceiverConnectionHandler).
-    /// See the comment on [`WalReceiverManager`](Self::WalReceiverManager).
+    /// Spawned by task [`WalReceiverConnectionHandler`].
+    /// See the comment on [`WalReceiverManager`].
     WalReceiverConnectionPoller,
 
     // Garbage collection worker. One per tenant

pageserver/src/tenant.rs

@@ -11,7 +11,7 @@
 //! parent timeline, and the last LSN that has been written to disk.
 //!
 
-use anyhow::{bail, Context};
+use anyhow::{bail, ensure, Context};
 use futures::FutureExt;
 use pageserver_api::models::TimelineState;
 use remote_storage::DownloadError;
@@ -49,8 +49,6 @@ use std::time::{Duration, Instant};
 use self::config::TenantConf;
 use self::metadata::TimelineMetadata;
 use self::remote_timeline_client::RemoteTimelineClient;
-use self::timeline::uninit::TimelineUninitMark;
-use self::timeline::uninit::UninitializedTimeline;
 use self::timeline::EvictionTaskTenantState;
 use crate::config::PageServerConf;
 use crate::context::{DownloadBehavior, RequestContext};
@@ -70,7 +68,6 @@ use crate::tenant::storage_layer::ImageLayer;
 use crate::tenant::storage_layer::Layer;
 use crate::InitializationOrder;
 
-use crate::tenant::timeline::uninit::cleanup_timeline_directory;
 use crate::virtual_file::VirtualFile;
 use crate::walredo::PostgresRedoManager;
 use crate::walredo::WalRedoManager;
@@ -84,32 +81,12 @@ use utils::{
     lsn::{Lsn, RecordLsn},
 };
 
-/// Declare a failpoint that can use the `pause` failpoint action.
-/// We don't want to block the executor thread, hence, spawn_blocking + await.
-macro_rules! pausable_failpoint {
-    ($name:literal) => {
-        if cfg!(feature = "testing") {
-            tokio::task::spawn_blocking({
-                let current = tracing::Span::current();
-                move || {
-                    let _entered = current.entered();
-                    tracing::info!("at failpoint {}", $name);
-                    fail::fail_point!($name);
-                }
-            })
-            .await
-            .expect("spawn_blocking");
-        }
-    };
-}
-
 pub mod blob_io;
 pub mod block_io;
 pub mod disk_btree;
 pub(crate) mod ephemeral_file;
 pub mod layer_map;
 pub mod manifest;
-mod span;
 
 pub mod metadata;
 mod par_fsync;
@@ -125,7 +102,7 @@ mod timeline;
 
 pub mod size;
 
-pub(crate) use timeline::span::debug_assert_current_span_has_tenant_and_timeline_id;
+pub(crate) use timeline::debug_assert_current_span_has_tenant_and_timeline_id;
 pub use timeline::{
     LocalLayerInfoForDiskUsageEviction, LogicalSizeCalculationCause, PageReconstructError, Timeline,
 };
@@ -133,7 +110,7 @@ pub use timeline::{
 // re-export this function so that page_cache.rs can use it.
 pub use crate::tenant::ephemeral_file::writeback as writeback_ephemeral_file;
 
-// re-export for use in remote_timeline_client.rs
+// re-export for use in storage_sync.rs
 pub use crate::tenant::metadata::save_metadata;
 
 // re-export for use in walreceiver
@@ -184,6 +161,200 @@ pub struct Tenant {
     eviction_task_tenant_state: tokio::sync::Mutex<EvictionTaskTenantState>,
 }
 
+/// A timeline with some of its files on disk, being initialized.
+/// This struct ensures the atomicity of the timeline init: it's either properly created and inserted into pageserver's memory, or
+/// its local files are removed. In the worst case of a crash, an uninit mark file is left behind, which causes the directory
+/// to be removed on next restart.
+///
+/// The caller is responsible for proper timeline data filling before the final init.
+#[must_use]
+pub struct UninitializedTimeline<'t> {
+    owning_tenant: &'t Tenant,
+    timeline_id: TimelineId,
+    raw_timeline: Option<(Arc<Timeline>, TimelineUninitMark)>,
+}
+
+/// An uninit mark file, created along the timeline dir to ensure the timeline either gets fully initialized and loaded into pageserver's memory,
+/// or gets removed eventually.
+///
+/// XXX: it's important to create it near the timeline dir, not inside it to ensure timeline dir gets removed first.
+#[must_use]
+struct TimelineUninitMark {
+    uninit_mark_deleted: bool,
+    uninit_mark_path: PathBuf,
+    timeline_path: PathBuf,
+}
+
+impl UninitializedTimeline<'_> {
+    /// Finish timeline creation: insert it into the Tenant's timelines map and remove the
+    /// uninit mark file.
+    ///
+    /// This function launches the flush loop if not already done.
+    ///
+    /// The caller is responsible for activating the timeline (function `.activate()`).
+    fn finish_creation(mut self) -> anyhow::Result<Arc<Timeline>> {
+        let timeline_id = self.timeline_id;
+        let tenant_id = self.owning_tenant.tenant_id;
+
+        let (new_timeline, uninit_mark) = self.raw_timeline.take().with_context(|| {
+            format!("No timeline for initalization found for {tenant_id}/{timeline_id}")
+        })?;
+
+        // Check that the caller initialized disk_consistent_lsn
+        let new_disk_consistent_lsn = new_timeline.get_disk_consistent_lsn();
+        ensure!(
+            new_disk_consistent_lsn.is_valid(),
+            "new timeline {tenant_id}/{timeline_id} has invalid disk_consistent_lsn"
+        );
+
+        let mut timelines = self.owning_tenant.timelines.lock().unwrap();
+        match timelines.entry(timeline_id) {
+            Entry::Occupied(_) => anyhow::bail!(
+                "Found freshly initialized timeline {tenant_id}/{timeline_id} in the tenant map"
+            ),
+            Entry::Vacant(v) => {
+                uninit_mark.remove_uninit_mark().with_context(|| {
+                    format!(
+                        "Failed to remove uninit mark file for timeline {tenant_id}/{timeline_id}"
+                    )
+                })?;
+                v.insert(Arc::clone(&new_timeline));
+
+                new_timeline.maybe_spawn_flush_loop();
+            }
+        }
+
+        Ok(new_timeline)
+    }
+
+    /// Prepares timeline data by loading it from the basebackup archive.
+    pub async fn import_basebackup_from_tar(
+        self,
+        copyin_read: &mut (impl tokio::io::AsyncRead + Send + Sync + Unpin),
+        base_lsn: Lsn,
+        broker_client: storage_broker::BrokerClientChannel,
+        ctx: &RequestContext,
+    ) -> anyhow::Result<Arc<Timeline>> {
+        let raw_timeline = self.raw_timeline()?;
+
+        import_datadir::import_basebackup_from_tar(raw_timeline, copyin_read, base_lsn, ctx)
+            .await
+            .context("Failed to import basebackup")?;
+
+        // Flush the new layer files to disk, before we make the timeline as available to
+        // the outside world.
+        //
+        // Flush loop needs to be spawned in order to be able to flush.
+        raw_timeline.maybe_spawn_flush_loop();
+
+        fail::fail_point!("before-checkpoint-new-timeline", |_| {
+            bail!("failpoint before-checkpoint-new-timeline");
+        });
+
+        raw_timeline
+            .freeze_and_flush()
+            .await
+            .context("Failed to flush after basebackup import")?;
+
+        // All the data has been imported. Insert the Timeline into the tenant's timelines
+        // map and remove the uninit mark file.
+        let tl = self.finish_creation()?;
+        tl.activate(broker_client, None, ctx);
+        Ok(tl)
+    }
+
+    fn raw_timeline(&self) -> anyhow::Result<&Arc<Timeline>> {
+        Ok(&self
+            .raw_timeline
+            .as_ref()
+            .with_context(|| {
+                format!(
+                    "No raw timeline {}/{} found",
+                    self.owning_tenant.tenant_id, self.timeline_id
+                )
+            })?
+            .0)
+    }
+}
+
+impl Drop for UninitializedTimeline<'_> {
+    fn drop(&mut self) {
+        if let Some((_, uninit_mark)) = self.raw_timeline.take() {
+            let _entered = info_span!("drop_uninitialized_timeline", tenant = %self.owning_tenant.tenant_id, timeline = %self.timeline_id).entered();
+            error!("Timeline got dropped without initializing, cleaning its files");
+            cleanup_timeline_directory(uninit_mark);
+        }
+    }
+}
+
+fn cleanup_timeline_directory(uninit_mark: TimelineUninitMark) {
+    let timeline_path = &uninit_mark.timeline_path;
+    match ignore_absent_files(|| fs::remove_dir_all(timeline_path)) {
+        Ok(()) => {
+            info!("Timeline dir {timeline_path:?} removed successfully, removing the uninit mark")
+        }
+        Err(e) => {
+            error!("Failed to clean up uninitialized timeline directory {timeline_path:?}: {e:?}")
+        }
+    }
+    drop(uninit_mark); // mark handles its deletion on drop, gets retained if timeline dir exists
+}
+
+impl TimelineUninitMark {
+    fn new(uninit_mark_path: PathBuf, timeline_path: PathBuf) -> Self {
+        Self {
+            uninit_mark_deleted: false,
+            uninit_mark_path,
+            timeline_path,
+        }
+    }
+
+    fn remove_uninit_mark(mut self) -> anyhow::Result<()> {
+        if !self.uninit_mark_deleted {
+            self.delete_mark_file_if_present()?;
+        }
+
+        Ok(())
+    }
+
+    fn delete_mark_file_if_present(&mut self) -> anyhow::Result<()> {
+        let uninit_mark_file = &self.uninit_mark_path;
+        let uninit_mark_parent = uninit_mark_file
+            .parent()
+            .with_context(|| format!("Uninit mark file {uninit_mark_file:?} has no parent"))?;
+        ignore_absent_files(|| fs::remove_file(uninit_mark_file)).with_context(|| {
+            format!("Failed to remove uninit mark file at path {uninit_mark_file:?}")
+        })?;
+        crashsafe::fsync(uninit_mark_parent).context("Failed to fsync uninit mark parent")?;
+        self.uninit_mark_deleted = true;
+
+        Ok(())
+    }
+}
+
+impl Drop for TimelineUninitMark {
+    fn drop(&mut self) {
+        if !self.uninit_mark_deleted {
+            if self.timeline_path.exists() {
+                error!(
+                    "Uninit mark {} is not removed, timeline {} stays uninitialized",
+                    self.uninit_mark_path.display(),
+                    self.timeline_path.display()
+                )
+            } else {
+                // unblock later timeline creation attempts
+                warn!(
+                    "Removing intermediate uninit mark file {}",
+                    self.uninit_mark_path.display()
+                );
+                if let Err(e) = self.delete_mark_file_if_present() {
+                    error!("Failed to remove the uninit mark file: {e}")
+                }
+            }
+        }
+    }
+}
+
 // We should not blindly overwrite local metadata with remote one.
 // For example, consider the following case:
 //     Image layer is flushed to disk as a new delta layer, we update local metadata and start upload task but after that
@@ -429,7 +600,7 @@ impl Tenant {
                     .layers
                     .read()
                     .await
-                    .layer_map()
+                    .0
                     .iter_historic_layers()
                     .next()
                     .is_some(),
@@ -440,8 +611,8 @@ impl Tenant {
         if !picked_local {
             save_metadata(
                 self.conf,
-                &tenant_id,
-                &timeline_id,
+                timeline_id,
+                tenant_id,
                 up_to_date_metadata,
                 first_save,
             )
@@ -470,7 +641,7 @@ impl Tenant {
     ) -> anyhow::Result<Arc<Tenant>> {
         // TODO dedup with spawn_load
         let tenant_conf =
-            Self::load_tenant_config(conf, &tenant_id).context("load tenant config")?;
+            Self::load_tenant_config(conf, tenant_id).context("load tenant config")?;
 
         let wal_redo_manager = Arc::new(PostgresRedoManager::new(conf, tenant_id));
         let tenant = Arc::new(Tenant::new(
@@ -524,7 +695,7 @@ impl Tenant {
     /// No background tasks are started as part of this routine.
     ///
     async fn attach(self: &Arc<Tenant>, ctx: &RequestContext) -> anyhow::Result<()> {
-        span::debug_assert_current_span_has_tenant_id();
+        debug_assert_current_span_has_tenant_id();
 
         let marker_file = self.conf.tenant_attaching_mark_file_path(&self.tenant_id);
         if !tokio::fs::try_exists(&marker_file)
@@ -579,7 +750,7 @@ impl Tenant {
                 .map(move |res| {
                     res.with_context(|| format!("download index part for timeline {timeline_id}"))
                 })
-                .instrument(info_span!("download_index_part", %timeline_id)),
+                .instrument(info_span!("download_index_part", timeline=%timeline_id)),
             );
         }
         // Wait for all the download tasks to complete & collect results.
@@ -662,10 +833,10 @@ impl Tenant {
         remote_client: RemoteTimelineClient,
         ctx: &RequestContext,
     ) -> anyhow::Result<()> {
-        span::debug_assert_current_span_has_tenant_id();
+        debug_assert_current_span_has_tenant_id();
 
         info!("downloading index file for timeline {}", timeline_id);
-        tokio::fs::create_dir_all(self.conf.timeline_path(&self.tenant_id, &timeline_id))
+        tokio::fs::create_dir_all(self.conf.timeline_path(&timeline_id, &self.tenant_id))
             .await
             .context("Failed to create new timeline directory")?;
 
@@ -741,9 +912,9 @@ impl Tenant {
         init_order: Option<InitializationOrder>,
         ctx: &RequestContext,
     ) -> Arc<Tenant> {
-        span::debug_assert_current_span_has_tenant_id();
+        debug_assert_current_span_has_tenant_id();
 
-        let tenant_conf = match Self::load_tenant_config(conf, &tenant_id) {
+        let tenant_conf = match Self::load_tenant_config(conf, tenant_id) {
             Ok(conf) => conf,
             Err(e) => {
                 error!("load tenant config failed: {:?}", e);
@@ -854,7 +1025,7 @@ impl Tenant {
                             timeline_uninit_mark_file.display()
                         )
                     })?;
-                let timeline_dir = self.conf.timeline_path(&self.tenant_id, &timeline_id);
+                let timeline_dir = self.conf.timeline_path(&timeline_id, &self.tenant_id);
                 if let Err(e) =
                     remove_timeline_and_uninit_mark(&timeline_dir, timeline_uninit_mark_file)
                 {
@@ -899,7 +1070,7 @@ impl Tenant {
                 if let Ok(timeline_id) =
                     file_name.to_str().unwrap_or_default().parse::<TimelineId>()
                 {
-                    let metadata = load_metadata(self.conf, &self.tenant_id, &timeline_id)
+                    let metadata = load_metadata(self.conf, timeline_id, self.tenant_id)
                         .context("failed to load metadata")?;
                     timelines_to_load.insert(timeline_id, metadata);
                 } else {
@@ -927,7 +1098,7 @@ impl Tenant {
         init_order: Option<&InitializationOrder>,
         ctx: &RequestContext,
     ) -> anyhow::Result<()> {
-        span::debug_assert_current_span_has_tenant_id();
+        debug_assert_current_span_has_tenant_id();
 
         debug!("loading tenant task");
 
@@ -973,7 +1144,7 @@ impl Tenant {
         init_order: Option<&InitializationOrder>,
         ctx: &RequestContext,
     ) -> anyhow::Result<()> {
-        span::debug_assert_current_span_has_tenant_id();
+        debug_assert_current_span_has_tenant_id();
 
         let remote_client = self.remote_storage.as_ref().map(|remote_storage| {
             RemoteTimelineClient::new(
@@ -1368,7 +1539,7 @@ impl Tenant {
         for (timeline_id, timeline) in &timelines_to_compact {
             timeline
                 .compact(ctx)
-                .instrument(info_span!("compact_timeline", %timeline_id))
+                .instrument(info_span!("compact_timeline", timeline = %timeline_id))
                 .await?;
         }
 
@@ -1459,12 +1630,12 @@ impl Tenant {
             let layer_removal_guard = timeline.layer_removal_cs.lock().await;
             info!("got layer_removal_cs.lock(), deleting layer files");
 
-            // NB: remote_timeline_client upload tasks that reference these layers have been cancelled
+            // NB: storage_sync upload tasks that reference these layers have been cancelled
             //     by the caller.
 
             let local_timeline_directory = self
                 .conf
-                .timeline_path(&self.tenant_id, &timeline.timeline_id);
+                .timeline_path(&timeline.timeline_id, &self.tenant_id);
 
             fail::fail_point!("timeline-delete-before-rm", |_| {
                 Err(anyhow::anyhow!("failpoint: timeline-delete-before-rm"))?
@@ -1517,7 +1688,20 @@ impl Tenant {
             remote_client.delete_all().await.context("delete_all")?
         };
 
-        pausable_failpoint!("in_progress_delete");
+        // Have a failpoint that can use the `pause` failpoint action.
+        // We don't want to block the executor thread, hence, spawn_blocking + await.
+        if cfg!(feature = "testing") {
+            tokio::task::spawn_blocking({
+                let current = tracing::Span::current();
+                move || {
+                    let _entered = current.entered();
+                    tracing::info!("at failpoint in_progress_delete");
+                    fail::fail_point!("in_progress_delete");
+                }
+            })
+            .await
+            .expect("spawn_blocking");
+        }
 
         {
             // Remove the timeline from the map.
@@ -1551,7 +1735,7 @@ impl Tenant {
         timeline_id: TimelineId,
         _ctx: &RequestContext,
     ) -> Result<(), DeleteTimelineError> {
-        debug_assert_current_span_has_tenant_and_timeline_id();
+        timeline::debug_assert_current_span_has_tenant_and_timeline_id();
 
         // Transition the timeline into TimelineState::Stopping.
         // This should prevent new operations from starting.
@@ -1715,7 +1899,7 @@ impl Tenant {
         background_jobs_can_start: Option<&completion::Barrier>,
         ctx: &RequestContext,
     ) {
-        span::debug_assert_current_span_has_tenant_id();
+        debug_assert_current_span_has_tenant_id();
 
         let mut activating = false;
         self.state.send_modify(|current_state| {
@@ -1786,7 +1970,7 @@ impl Tenant {
     ///
     /// This will attempt to shutdown even if tenant is broken.
     pub(crate) async fn shutdown(&self, freeze_and_flush: bool) -> Result<(), ShutdownError> {
-        span::debug_assert_current_span_has_tenant_id();
+        debug_assert_current_span_has_tenant_id();
         // Set tenant (and its timlines) to Stoppping state.
         //
         // Since we can only transition into Stopping state after activation is complete,
@@ -2232,7 +2416,7 @@ impl Tenant {
     /// Locate and load config
     pub(super) fn load_tenant_config(
         conf: &'static PageServerConf,
-        tenant_id: &TenantId,
+        tenant_id: TenantId,
     ) -> anyhow::Result<TenantConfOpt> {
         let target_config_path = conf.tenant_config_path(tenant_id);
         let target_config_display = target_config_path.display();
@@ -2819,7 +3003,7 @@ impl Tenant {
         timeline_struct.init_empty_layer_map(start_lsn);
 
         if let Err(e) =
-            self.create_timeline_files(&uninit_mark.timeline_path, &new_timeline_id, new_metadata)
+            self.create_timeline_files(&uninit_mark.timeline_path, new_timeline_id, new_metadata)
         {
             error!("Failed to create initial files for timeline {tenant_id}/{new_timeline_id}, cleaning up: {e:?}");
             cleanup_timeline_directory(uninit_mark);
@@ -2828,17 +3012,17 @@ impl Tenant {
 
         debug!("Successfully created initial files for timeline {tenant_id}/{new_timeline_id}");
 
-        Ok(UninitializedTimeline::new(
-            self,
-            new_timeline_id,
-            Some((timeline_struct, uninit_mark)),
-        ))
+        Ok(UninitializedTimeline {
+            owning_tenant: self,
+            timeline_id: new_timeline_id,
+            raw_timeline: Some((timeline_struct, uninit_mark)),
+        })
     }
 
     fn create_timeline_files(
         &self,
         timeline_path: &Path,
-        new_timeline_id: &TimelineId,
+        new_timeline_id: TimelineId,
         new_metadata: &TimelineMetadata,
     ) -> anyhow::Result<()> {
         crashsafe::create_dir(timeline_path).context("Failed to create timeline directory")?;
@@ -2849,8 +3033,8 @@ impl Tenant {
 
         save_metadata(
             self.conf,
-            &self.tenant_id,
             new_timeline_id,
+            self.tenant_id,
             new_metadata,
             true,
         )
@@ -2873,7 +3057,7 @@ impl Tenant {
             timelines.get(&timeline_id).is_none(),
             "Timeline {tenant_id}/{timeline_id} already exists in pageserver's memory"
         );
-        let timeline_path = self.conf.timeline_path(&tenant_id, &timeline_id);
+        let timeline_path = self.conf.timeline_path(&timeline_id, &tenant_id);
         anyhow::ensure!(
             !timeline_path.exists(),
             "Timeline {} already exists, cannot create its uninit mark file",
@@ -3004,10 +3188,10 @@ pub(crate) enum CreateTenantFilesMode {
 pub(crate) fn create_tenant_files(
     conf: &'static PageServerConf,
     tenant_conf: TenantConfOpt,
-    tenant_id: &TenantId,
+    tenant_id: TenantId,
     mode: CreateTenantFilesMode,
 ) -> anyhow::Result<PathBuf> {
-    let target_tenant_directory = conf.tenant_path(tenant_id);
+    let target_tenant_directory = conf.tenant_path(&tenant_id);
     anyhow::ensure!(
         !target_tenant_directory
             .try_exists()
@@ -3058,7 +3242,7 @@ pub(crate) fn create_tenant_files(
 fn try_create_target_tenant_dir(
     conf: &'static PageServerConf,
     tenant_conf: TenantConfOpt,
-    tenant_id: &TenantId,
+    tenant_id: TenantId,
     mode: CreateTenantFilesMode,
     temporary_tenant_dir: &Path,
     target_tenant_directory: &Path,
@@ -3082,7 +3266,7 @@ fn try_create_target_tenant_dir(
     }
 
     let temporary_tenant_timelines_dir = rebase_directory(
-        &conf.timelines_path(tenant_id),
+        &conf.timelines_path(&tenant_id),
         target_tenant_directory,
         temporary_tenant_dir,
     )
@@ -3094,7 +3278,7 @@ fn try_create_target_tenant_dir(
     )
     .with_context(|| format!("resolve tenant {tenant_id} temporary config path"))?;
 
-    Tenant::persist_tenant_config(tenant_id, &temporary_tenant_config_path, tenant_conf, true)?;
+    Tenant::persist_tenant_config(&tenant_id, &temporary_tenant_config_path, tenant_conf, true)?;
 
     crashsafe::create_dir(&temporary_tenant_timelines_dir).with_context(|| {
         format!(
@@ -3382,7 +3566,7 @@ pub mod harness {
         }
 
         pub fn timeline_path(&self, timeline_id: &TimelineId) -> PathBuf {
-            self.conf.timeline_path(&self.tenant_id, timeline_id)
+            self.conf.timeline_path(timeline_id, &self.tenant_id)
         }
     }
 
@@ -4335,13 +4519,13 @@ mod tests {
         // assert freeze_and_flush exercised the initdb optimization
         {
             let state = tline.flush_loop_state.lock().unwrap();
-            let timeline::FlushLoopState::Running {
-                expect_initdb_optimization,
-                initdb_optimization_count,
-            } = *state
-            else {
-                panic!("unexpected state: {:?}", *state);
-            };
+            let
+                timeline::FlushLoopState::Running {
+                    expect_initdb_optimization,
+                    initdb_optimization_count,
+                } = *state else {
+                    panic!("unexpected state: {:?}", *state);
+                };
             assert!(expect_initdb_optimization);
             assert!(initdb_optimization_count > 0);
         }
@@ -4376,7 +4560,7 @@ mod tests {
 
         assert!(!harness
             .conf
-            .timeline_path(&tenant.tenant_id, &TIMELINE_ID)
+            .timeline_path(&TIMELINE_ID, &tenant.tenant_id)
             .exists());
 
         assert!(!harness
@@ -4387,3 +4571,28 @@ mod tests {
         Ok(())
     }
 }
+
+#[cfg(not(debug_assertions))]
+#[inline]
+pub(crate) fn debug_assert_current_span_has_tenant_id() {}
+
+#[cfg(debug_assertions)]
+pub static TENANT_ID_EXTRACTOR: once_cell::sync::Lazy<
+    utils::tracing_span_assert::MultiNameExtractor<2>,
+> = once_cell::sync::Lazy::new(|| {
+    utils::tracing_span_assert::MultiNameExtractor::new("TenantId", ["tenant_id", "tenant"])
+});
+
+#[cfg(debug_assertions)]
+#[inline]
+pub(crate) fn debug_assert_current_span_has_tenant_id() {
+    use utils::tracing_span_assert;
+
+    match tracing_span_assert::check_fields_present([&*TENANT_ID_EXTRACTOR]) {
+        Ok(()) => (),
+        Err(missing) => panic!(
+            "missing extractors: {:?}",
+            missing.into_iter().map(|e| e.name()).collect::<Vec<_>>()
+        ),
+    }
+}

pageserver/src/tenant/disk_btree.rs

@@ -442,7 +442,7 @@ where
     writer: W,
 
     ///
-    /// `stack[0]` is the current root page, `stack.last()` is the leaf.
+    /// stack[0] is the current root page, stack.last() is the leaf.
     ///
     /// We maintain the length of the stack to be always greater than zero.
     /// Two exceptions are:

pageserver/src/tenant/ephemeral_file.rs

@@ -55,7 +55,7 @@ impl EphemeralFile {
         l.next_file_id += 1;
 
         let filename = conf
-            .timeline_path(&tenant_id, &timeline_id)
+            .timeline_path(&timeline_id, &tenant_id)
             .join(PathBuf::from(format!("ephemeral-{}", file_id)));
 
         let file = VirtualFile::open_with_options(
@@ -346,7 +346,7 @@ mod tests {
 
         let tenant_id = TenantId::from_str("11000000000000000000000000000000").unwrap();
         let timeline_id = TimelineId::from_str("22000000000000000000000000000000").unwrap();
-        fs::create_dir_all(conf.timeline_path(&tenant_id, &timeline_id))?;
+        fs::create_dir_all(conf.timeline_path(&timeline_id, &tenant_id))?;
 
         Ok((conf, tenant_id, timeline_id))
     }

pageserver/src/tenant/layer_map.rs

@@ -16,7 +16,7 @@
 //! Other read methods are less critical but still impact performance of background tasks.
 //!
 //! This data structure relies on a persistent/immutable binary search tree. See the
-//! following lecture for an introduction <https://www.youtube.com/watch?v=WqCWghETNDc&t=581s>
+//! following lecture for an introduction https://www.youtube.com/watch?v=WqCWghETNDc&t=581s
 //! Summary: A persistent/immutable BST (and persistent data structures in general) allows
 //! you to modify the tree in such a way that each modification creates a new "version"
 //! of the tree. When you modify it, you get a new version, but all previous versions are
@@ -40,7 +40,7 @@
 //! afterwards. We can add layers as long as they have larger LSNs than any previous layer in
 //! the map, but if we need to remove a layer, or insert anything with an older LSN, we need
 //! to throw away most of the persistent BST and build a new one, starting from the oldest
-//! LSN. See [`LayerMap::flush_updates()`].
+//! LSN. See `LayerMap::flush_updates()`.
 //!
 
 mod historic_layer_coverage;
@@ -60,6 +60,7 @@ use utils::lsn::Lsn;
 use historic_layer_coverage::BufferedHistoricLayerCoverage;
 pub use historic_layer_coverage::LayerKey;
 
+use super::storage_layer::range_eq;
 use super::storage_layer::PersistentLayerDesc;
 
 ///
@@ -364,7 +365,7 @@ impl LayerMap {
     }
 
     pub fn is_l0(layer: &PersistentLayerDesc) -> bool {
-        layer.get_key_range() == (Key::MIN..Key::MAX)
+        range_eq(&layer.get_key_range(), &(Key::MIN..Key::MAX))
     }
 
     /// This function determines which layers are counted in `count_deltas`:
@@ -396,7 +397,7 @@ impl LayerMap {
         }
 
         // Case 2
-        if partition_range == &(Key::MIN..Key::MAX) {
+        if range_eq(partition_range, &(Key::MIN..Key::MAX)) {
             return true;
         }
 
@@ -651,35 +652,19 @@ impl LayerMap {
 #[cfg(test)]
 mod tests {
     use super::LayerMap;
-    use crate::tenant::storage_layer::LayerFileName;
+    use crate::tenant::storage_layer::{tests::LayerDescriptor, LayerFileName};
     use std::str::FromStr;
     use std::sync::Arc;
 
     mod l0_delta_layers_updated {
 
         use crate::tenant::{
-            storage_layer::{AsLayerDesc, PersistentLayerDesc},
-            timeline::layer_manager::LayerFileManager,
+            storage_layer::{PersistentLayer, PersistentLayerDesc},
+            timeline::LayerFileManager,
         };
 
         use super::*;
 
-        struct LayerObject(PersistentLayerDesc);
-
-        impl AsLayerDesc for LayerObject {
-            fn layer_desc(&self) -> &PersistentLayerDesc {
-                &self.0
-            }
-        }
-
-        impl LayerObject {
-            fn new(desc: PersistentLayerDesc) -> Self {
-                LayerObject(desc)
-            }
-        }
-
-        type TestLayerFileManager = LayerFileManager<LayerObject>;
-
         #[test]
         fn for_full_range_delta() {
             // l0_delta_layers are used by compaction, and should observe all buffered updates
@@ -716,18 +701,18 @@ mod tests {
 
             let layer = "000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__0000000053423C21-0000000053424D69";
             let layer = LayerFileName::from_str(layer).unwrap();
-            let layer = PersistentLayerDesc::from(layer);
+            let layer = LayerDescriptor::from(layer);
 
             // same skeletan construction; see scenario below
-            let not_found = Arc::new(LayerObject::new(layer.clone()));
-            let new_version = Arc::new(LayerObject::new(layer));
+            let not_found = Arc::new(layer.clone());
+            let new_version = Arc::new(layer);
 
             // after the immutable storage state refactor, the replace operation
             // will not use layer map any more. We keep it here for consistency in test cases
             // and can remove it in the future.
             let _map = LayerMap::default();
 
-            let mut mapping = TestLayerFileManager::new();
+            let mut mapping = LayerFileManager::new();
 
             mapping
                 .replace_and_verify(not_found, new_version)
@@ -736,10 +721,10 @@ mod tests {
 
         fn l0_delta_layers_updated_scenario(layer_name: &str, expected_l0: bool) {
             let name = LayerFileName::from_str(layer_name).unwrap();
-            let skeleton = PersistentLayerDesc::from(name);
+            let skeleton = LayerDescriptor::from(name);
 
-            let remote = Arc::new(LayerObject::new(skeleton.clone()));
-            let downloaded = Arc::new(LayerObject::new(skeleton));
+            let remote = Arc::new(skeleton.clone());
+            let downloaded = Arc::new(skeleton);
 
             let mut map = LayerMap::default();
             let mut mapping = LayerFileManager::new();

pageserver/src/tenant/layer_map/historic_layer_coverage.rs

@@ -122,7 +122,8 @@ impl<Value: Clone> HistoricLayerCoverage<Value> {
         self.head = self
             .historic
             .iter()
-            .next_back()
+            .rev()
+            .next()
             .map(|(_, v)| v.clone())
             .unwrap_or_default();
     }
@@ -411,7 +412,7 @@ fn test_persistent_overlapping() {
 /// still be more critical.
 ///
 /// See this for more on persistent and retroactive techniques:
-/// <https://www.youtube.com/watch?v=WqCWghETNDc&t=581s>
+/// https://www.youtube.com/watch?v=WqCWghETNDc&t=581s
 pub struct BufferedHistoricLayerCoverage<Value> {
     /// A persistent layer map that we rebuild when we need to retroactively update
     historic_coverage: HistoricLayerCoverage<Value>,

pageserver/src/tenant/layer_map/layer_coverage.rs

@@ -2,7 +2,7 @@ use std::ops::Range;
 
 // NOTE the `im` crate has 20x more downloads and also has
 // persistent/immutable BTree. But it's bugged so rpds is a
-// better choice <https://github.com/neondatabase/neon/issues/3395>
+// better choice https://github.com/neondatabase/neon/issues/3395
 use rpds::RedBlackTreeMapSync;
 
 /// Data structure that can efficiently:
@@ -11,7 +11,7 @@ use rpds::RedBlackTreeMapSync;
 /// - insert layers in non-decreasing lsn.start order
 ///
 /// For a detailed explanation and justification of this approach, see:
-/// <https://neon.tech/blog/persistent-structures-in-neons-wal-indexing>
+/// https://neon.tech/blog/persistent-structures-in-neons-wal-indexing
 ///
 /// NOTE The struct is parameterized over Value for easier
 ///      testing, but in practice it's some sort of layer.
@@ -113,7 +113,8 @@ impl<Value: Clone> LayerCoverage<Value> {
     pub fn query(&self, key: i128) -> Option<Value> {
         self.nodes
             .range(..=key)
-            .next_back()?
+            .rev()
+            .next()?
             .1
             .as_ref()
             .map(|(_, v)| v.clone())

pageserver/src/tenant/manifest.rs

@@ -24,7 +24,7 @@
 //! Currently, this is not used in the system. Future refactors will ensure
 //! the storage state will be recorded in this file, and the system can be
 //! recovered from this file. This is tracked in
-//! <https://github.com/neondatabase/neon/issues/4418>
+//! https://github.com/neondatabase/neon/issues/4418
 
 use std::io::{self, Read, Write};
 

pageserver/src/tenant/metadata.rs

@@ -1,12 +1,10 @@
 //! Every image of a certain timeline from [`crate::tenant::Tenant`]
 //! has a metadata that needs to be stored persistently.
 //!
-//! Later, the file gets used in [`remote_timeline_client`] as a part of
+//! Later, the file gets is used in [`crate::remote_storage::storage_sync`] as a part of
 //! external storage import and export operations.
 //!
 //! The module contains all structs and related helper methods related to timeline metadata.
-//!
-//! [`remote_timeline_client`]: super::remote_timeline_client
 
 use std::fs::{File, OpenOptions};
 use std::io::Write;
@@ -234,13 +232,13 @@ impl TimelineMetadata {
 /// Save timeline metadata to file
 pub fn save_metadata(
     conf: &'static PageServerConf,
-    tenant_id: &TenantId,
-    timeline_id: &TimelineId,
+    timeline_id: TimelineId,
+    tenant_id: TenantId,
     data: &TimelineMetadata,
     first_save: bool,
 ) -> anyhow::Result<()> {
     let _enter = info_span!("saving metadata").entered();
-    let path = conf.metadata_path(tenant_id, timeline_id);
+    let path = conf.metadata_path(timeline_id, tenant_id);
     // use OpenOptions to ensure file presence is consistent with first_save
     let mut file = VirtualFile::open_with_options(
         &path,
@@ -269,10 +267,10 @@ pub fn save_metadata(
 
 pub fn load_metadata(
     conf: &'static PageServerConf,
-    tenant_id: &TenantId,
-    timeline_id: &TimelineId,
+    timeline_id: TimelineId,
+    tenant_id: TenantId,
 ) -> anyhow::Result<TimelineMetadata> {
-    let metadata_path = conf.metadata_path(tenant_id, timeline_id);
+    let metadata_path = conf.metadata_path(timeline_id, tenant_id);
     let metadata_bytes = std::fs::read(&metadata_path).with_context(|| {
         format!(
             "Failed to read metadata bytes from path {}",

pageserver/src/tenant/mgr.rs

@@ -184,9 +184,9 @@ pub fn schedule_local_tenant_processing(
             format!("Could not parse tenant id out of the tenant dir name in path {tenant_path:?}")
         })?;
 
-    let tenant_ignore_mark = conf.tenant_ignore_mark_file_path(&tenant_id);
+    let tenant_ignore_mark = conf.tenant_ignore_mark_file_path(tenant_id);
     anyhow::ensure!(
-        !conf.tenant_ignore_mark_file_path(&tenant_id).exists(),
+        !conf.tenant_ignore_mark_file_path(tenant_id).exists(),
         "Cannot load tenant, ignore mark found at {tenant_ignore_mark:?}"
     );
 
@@ -310,7 +310,7 @@ pub async fn create_tenant(
         // We're holding the tenants lock in write mode while doing local IO.
         // If this section ever becomes contentious, introduce a new `TenantState::Creating`
         // and do the work in that state.
-        let tenant_directory = super::create_tenant_files(conf, tenant_conf, &tenant_id, CreateTenantFilesMode::Create)?;
+        let tenant_directory = super::create_tenant_files(conf, tenant_conf, tenant_id, CreateTenantFilesMode::Create)?;
         // TODO: tenant directory remains on disk if we bail out from here on.
         //       See https://github.com/neondatabase/neon/issues/4233
 
@@ -344,9 +344,14 @@ pub async fn set_new_tenant_config(
     info!("configuring tenant {tenant_id}");
     let tenant = get_tenant(tenant_id, true).await?;
 
-    let tenant_config_path = conf.tenant_config_path(&tenant_id);
-    Tenant::persist_tenant_config(&tenant_id, &tenant_config_path, new_tenant_conf, false)
-        .map_err(SetNewTenantConfigError::Persist)?;
+    let tenant_config_path = conf.tenant_config_path(tenant_id);
+    Tenant::persist_tenant_config(
+        &tenant.tenant_id(),
+        &tenant_config_path,
+        new_tenant_conf,
+        false,
+    )
+    .map_err(SetNewTenantConfigError::Persist)?;
     tenant.set_new_tenant_config(new_tenant_conf);
     Ok(())
 }
@@ -430,7 +435,7 @@ pub async fn detach_tenant(
     // Ignored tenants are not present in memory and will bail the removal from memory operation.
     // Before returning the error, check for ignored tenant removal case — we only need to clean its local files then.
     if detach_ignored && matches!(removal_result, Err(TenantStateError::NotFound(_))) {
-        let tenant_ignore_mark = conf.tenant_ignore_mark_file_path(&tenant_id);
+        let tenant_ignore_mark = conf.tenant_ignore_mark_file_path(tenant_id);
         if tenant_ignore_mark.exists() {
             info!("Detaching an ignored tenant");
             local_files_cleanup_operation(tenant_id)
@@ -452,7 +457,7 @@ pub async fn load_tenant(
 ) -> Result<(), TenantMapInsertError> {
     tenant_map_insert(tenant_id, || {
         let tenant_path = conf.tenant_path(&tenant_id);
-        let tenant_ignore_mark = conf.tenant_ignore_mark_file_path(&tenant_id);
+        let tenant_ignore_mark = conf.tenant_ignore_mark_file_path(tenant_id);
         if tenant_ignore_mark.exists() {
             std::fs::remove_file(&tenant_ignore_mark)
                 .with_context(|| format!("Failed to remove tenant ignore mark {tenant_ignore_mark:?} during tenant loading"))?;
@@ -473,7 +478,7 @@ pub async fn ignore_tenant(
     tenant_id: TenantId,
 ) -> Result<(), TenantStateError> {
     remove_tenant_from_memory(tenant_id, async {
-        let ignore_mark_file = conf.tenant_ignore_mark_file_path(&tenant_id);
+        let ignore_mark_file = conf.tenant_ignore_mark_file_path(tenant_id);
         fs::File::create(&ignore_mark_file)
             .await
             .context("Failed to create ignore mark file")
@@ -520,7 +525,7 @@ pub async fn attach_tenant(
     ctx: &RequestContext,
 ) -> Result<(), TenantMapInsertError> {
     tenant_map_insert(tenant_id, || {
-        let tenant_dir = create_tenant_files(conf, tenant_conf, &tenant_id, CreateTenantFilesMode::Attach)?;
+        let tenant_dir = create_tenant_files(conf, tenant_conf, tenant_id, CreateTenantFilesMode::Attach)?;
         // TODO: tenant directory remains on disk if we bail out from here on.
         //       See https://github.com/neondatabase/neon/issues/4233
 
@@ -690,7 +695,7 @@ pub async fn immediate_gc(
             fail::fail_point!("immediate_gc_task_pre");
             let result = tenant
                 .gc_iteration(Some(timeline_id), gc_horizon, pitr, &ctx)
-                .instrument(info_span!("manual_gc", %tenant_id, %timeline_id))
+                .instrument(info_span!("manual_gc", tenant = %tenant_id, timeline = %timeline_id))
                 .await;
                 // FIXME: `gc_iteration` can return an error for multiple reasons; we should handle it
                 // better once the types support it.
@@ -740,7 +745,9 @@ pub async fn immediate_compact(
         async move {
             let result = timeline
                 .compact(&ctx)
-                .instrument(info_span!("manual_compact", %tenant_id, %timeline_id))
+                .instrument(
+                    info_span!("manual_compact", tenant = %tenant_id, timeline = %timeline_id),
+                )
                 .await;
 
             match task_done.send(result) {

pageserver/src/tenant/remote_timeline_client.rs

@@ -135,7 +135,7 @@
 //! - Initiate upload queue with that [`IndexPart`].
 //! - Reschedule all lost operations by comparing the local filesystem state
 //!   and remote state as per [`IndexPart`]. This is done in
-//!   [`Tenant::timeline_init_and_sync`] and [`Timeline::reconcile_with_remote`].
+//!   [`Timeline::timeline_init_and_sync`] and [`Timeline::reconcile_with_remote`].
 //!
 //! Note that if we crash during file deletion between the index update
 //! that removes the file from the list of files, and deleting the remote file,
@@ -163,8 +163,8 @@
 //!   - download their remote [`IndexPart`]s
 //!   - create `Timeline` struct and a `RemoteTimelineClient`
 //!   - initialize the client's upload queue with its `IndexPart`
-//!   - create [`RemoteLayer`](super::storage_layer::RemoteLayer) instances
-//!     for layers that are referenced by `IndexPart` but not present locally
+//!   - create [`RemoteLayer`] instances for layers that are referenced by `IndexPart`
+//!     but not present locally
 //!   - schedule uploads for layers that are only present locally.
 //!   - if the remote `IndexPart`'s metadata was newer than the metadata in
 //!     the local filesystem, write the remote metadata to the local filesystem
@@ -198,8 +198,6 @@
 //! in remote storage.
 //! But note that we don't test any of this right now.
 //!
-//! [`Tenant::timeline_init_and_sync`]: super::Tenant::timeline_init_and_sync
-//! [`Timeline::reconcile_with_remote`]: super::Timeline::reconcile_with_remote
 
 mod delete;
 mod download;
@@ -444,8 +442,8 @@ impl RemoteTimelineClient {
         let index_part = download::download_index_part(
             self.conf,
             &self.storage_impl,
-            &self.tenant_id,
-            &self.timeline_id,
+            self.tenant_id,
+            self.timeline_id,
         )
         .measure_remote_op(
             self.tenant_id,
@@ -750,13 +748,25 @@ impl RemoteTimelineClient {
             stopped.deleted_at = SetDeletedFlagProgress::NotRunning;
         });
 
-        pausable_failpoint!("persist_deleted_index_part");
-
+        // Have a failpoint that can use the `pause` failpoint action.
+        // We don't want to block the executor thread, hence, spawn_blocking + await.
+        if cfg!(feature = "testing") {
+            tokio::task::spawn_blocking({
+                let current = tracing::Span::current();
+                move || {
+                    let _entered = current.entered();
+                    tracing::info!("at failpoint persist_deleted_index_part");
+                    fail::fail_point!("persist_deleted_index_part");
+                }
+            })
+            .await
+            .expect("spawn_blocking");
+        }
         upload::upload_index_part(
             self.conf,
             &self.storage_impl,
-            &self.tenant_id,
-            &self.timeline_id,
+            self.tenant_id,
+            self.timeline_id,
             &index_part_with_deleted_at,
         )
         .await?;
@@ -831,7 +841,7 @@ impl RemoteTimelineClient {
 
         // Do not delete index part yet, it is needed for possible retry. If we remove it first
         // and retry will arrive to different pageserver there wont be any traces of it on remote storage
-        let timeline_path = self.conf.timeline_path(&self.tenant_id, &self.timeline_id);
+        let timeline_path = self.conf.timeline_path(&self.timeline_id, &self.tenant_id);
         let timeline_storage_path = self.conf.remote_path(&timeline_path)?;
 
         let remaining = self
@@ -842,16 +852,14 @@ impl RemoteTimelineClient {
         let remaining: Vec<RemotePath> = remaining
             .into_iter()
             .filter(|p| p.object_name() != Some(IndexPart::FILE_NAME))
-            .inspect(|path| {
-                if let Some(name) = path.object_name() {
-                    info!(%name, "deleting a file not referenced from index_part.json");
-                } else {
-                    warn!(%path, "deleting a nameless or non-utf8 object not referenced from index_part.json");
-                }
-            })
             .collect();
 
         if !remaining.is_empty() {
+            warn!(
+                "Found {} files not bound to index_file.json, proceeding with their deletion",
+                remaining.len()
+            );
+            warn!("About to remove {} files", remaining.len());
             self.storage_impl.delete_objects(&remaining).await?;
         }
 
@@ -860,7 +868,7 @@ impl RemoteTimelineClient {
         debug!("deleting index part");
         self.storage_impl.delete(&index_file_path).await?;
 
-        info!(prefix=%timeline_storage_path, referenced=deletions_queued, not_referenced=%remaining.len(), "done deleting in timeline prefix, including index_part.json");
+        info!(deletions_queued, "done deleting, including index_part.json");
 
         Ok(())
     }
@@ -925,11 +933,11 @@ impl RemoteTimelineClient {
 
             // Assign unique ID to this task
             upload_queue.task_counter += 1;
-            let upload_task_id = upload_queue.task_counter;
+            let task_id = upload_queue.task_counter;
 
             // Add it to the in-progress map
             let task = Arc::new(UploadTask {
-                task_id: upload_task_id,
+                task_id,
                 op: next_op,
                 retries: AtomicU32::new(0),
             });
@@ -939,8 +947,6 @@ impl RemoteTimelineClient {
 
             // Spawn task to perform the task
             let self_rc = Arc::clone(self);
-            let tenant_id = self.tenant_id;
-            let timeline_id = self.timeline_id;
             task_mgr::spawn(
                 self.runtime.handle(),
                 TaskKind::RemoteUploadTask,
@@ -952,7 +958,7 @@ impl RemoteTimelineClient {
                     self_rc.perform_upload_task(task).await;
                     Ok(())
                 }
-                .instrument(info_span!(parent: None, "remote_upload", %tenant_id, %timeline_id, %upload_task_id)),
+                .instrument(info_span!(parent: None, "remote_upload", tenant = %self.tenant_id, timeline = %self.timeline_id, upload_task_id = %task_id)),
             );
 
             // Loop back to process next task
@@ -997,7 +1003,7 @@ impl RemoteTimelineClient {
                 UploadOp::UploadLayer(ref layer_file_name, ref layer_metadata) => {
                     let path = &self
                         .conf
-                        .timeline_path(&self.tenant_id, &self.timeline_id)
+                        .timeline_path(&self.timeline_id, &self.tenant_id)
                         .join(layer_file_name.file_name());
                     upload::upload_timeline_layer(
                         self.conf,
@@ -1018,8 +1024,8 @@ impl RemoteTimelineClient {
                     let res = upload::upload_index_part(
                         self.conf,
                         &self.storage_impl,
-                        &self.tenant_id,
-                        &self.timeline_id,
+                        self.tenant_id,
+                        self.timeline_id,
                         index_part,
                     )
                     .measure_remote_op(
@@ -1038,7 +1044,7 @@ impl RemoteTimelineClient {
                 UploadOp::Delete(delete) => {
                     let path = &self
                         .conf
-                        .timeline_path(&self.tenant_id, &self.timeline_id)
+                        .timeline_path(&self.timeline_id, &self.tenant_id)
                         .join(delete.layer_file_name.file_name());
                     delete::delete_layer(self.conf, &self.storage_impl, path)
                         .measure_remote_op(

pageserver/src/tenant/remote_timeline_client/delete.rs

@@ -19,10 +19,9 @@ pub(super) async fn delete_layer<'a>(
 
     let path_to_delete = conf.remote_path(local_layer_path)?;
 
-    // We don't want to print an error if the delete failed if the file has
-    // already been deleted. Thankfully, in this situation S3 already
-    // does not yield an error. While OS-provided local file system APIs do yield
-    // errors, we avoid them in the `LocalFs` wrapper.
+    // XXX: If the deletion fails because the object already didn't exist,
+    // it would be good to just issue a warning but consider it success.
+    // https://github.com/neondatabase/neon/issues/2934
     storage.delete(&path_to_delete).await.with_context(|| {
         format!("Failed to delete remote layer from storage at {path_to_delete:?}")
     })

pageserver/src/tenant/remote_timeline_client/download.rs

@@ -16,7 +16,7 @@ use tracing::{info, warn};
 
 use crate::config::PageServerConf;
 use crate::tenant::storage_layer::LayerFileName;
-use crate::tenant::timeline::span::debug_assert_current_span_has_tenant_and_timeline_id;
+use crate::tenant::timeline::debug_assert_current_span_has_tenant_and_timeline_id;
 use crate::{exponential_backoff, DEFAULT_BASE_BACKOFF_SECONDS, DEFAULT_MAX_BACKOFF_SECONDS};
 use remote_storage::{DownloadError, GenericRemoteStorage};
 use utils::crashsafe::path_with_suffix_extension;
@@ -46,7 +46,7 @@ pub async fn download_layer_file<'a>(
 ) -> Result<u64, DownloadError> {
     debug_assert_current_span_has_tenant_and_timeline_id();
 
-    let timeline_path = conf.timeline_path(&tenant_id, &timeline_id);
+    let timeline_path = conf.timeline_path(&timeline_id, &tenant_id);
 
     let local_path = timeline_path.join(layer_file_name.file_name());
 
@@ -229,11 +229,11 @@ pub async fn list_remote_timelines<'a>(
 pub(super) async fn download_index_part(
     conf: &'static PageServerConf,
     storage: &GenericRemoteStorage,
-    tenant_id: &TenantId,
-    timeline_id: &TimelineId,
+    tenant_id: TenantId,
+    timeline_id: TimelineId,
 ) -> Result<IndexPart, DownloadError> {
     let index_part_path = conf
-        .metadata_path(tenant_id, timeline_id)
+        .metadata_path(timeline_id, tenant_id)
         .with_file_name(IndexPart::FILE_NAME);
     let part_storage_path = conf
         .remote_path(&index_part_path)

pageserver/src/tenant/remote_timeline_client/upload.rs

@@ -2,7 +2,7 @@
 
 use anyhow::{bail, Context};
 use fail::fail_point;
-use std::{io::ErrorKind, path::Path};
+use std::path::Path;
 use tokio::fs;
 
 use crate::{config::PageServerConf, tenant::remote_timeline_client::index::IndexPart};
@@ -11,14 +11,12 @@ use utils::id::{TenantId, TimelineId};
 
 use super::index::LayerFileMetadata;
 
-use tracing::info;
-
 /// Serializes and uploads the given index part data to the remote storage.
 pub(super) async fn upload_index_part<'a>(
     conf: &'static PageServerConf,
     storage: &'a GenericRemoteStorage,
-    tenant_id: &TenantId,
-    timeline_id: &TimelineId,
+    tenant_id: TenantId,
+    timeline_id: TimelineId,
     index_part: &'a IndexPart,
 ) -> anyhow::Result<()> {
     tracing::trace!("uploading new index part");
@@ -33,7 +31,7 @@ pub(super) async fn upload_index_part<'a>(
     let index_part_bytes = tokio::io::BufReader::new(std::io::Cursor::new(index_part_bytes));
 
     let index_part_path = conf
-        .metadata_path(tenant_id, timeline_id)
+        .metadata_path(timeline_id, tenant_id)
         .with_file_name(IndexPart::FILE_NAME);
     let storage_path = conf.remote_path(&index_part_path)?;
 
@@ -58,21 +56,9 @@ pub(super) async fn upload_timeline_layer<'a>(
     });
     let storage_path = conf.remote_path(source_path)?;
 
-    let source_file_res = fs::File::open(&source_path).await;
-    let source_file = match source_file_res {
-        Ok(source_file) => source_file,
-        Err(e) if e.kind() == ErrorKind::NotFound => {
-            // If we encounter this arm, it wasn't intended, but it's also not
-            // a big problem, if it's because the file was deleted before an
-            // upload. However, a nonexistent file can also be indicative of
-            // something worse, like when a file is scheduled for upload before
-            // it has been written to disk yet.
-            info!(path = %source_path.display(), "File to upload doesn't exist. Likely the file has been deleted and an upload is not required any more.");
-            return Ok(());
-        }
-        Err(e) => Err(e)
-            .with_context(|| format!("Failed to open a source file for layer {source_path:?}"))?,
-    };
+    let source_file = fs::File::open(&source_path)
+        .await
+        .with_context(|| format!("Failed to open a source file for layer {source_path:?}"))?;
 
     let fs_size = source_file
         .metadata()

pageserver/src/tenant/size.rs

@@ -110,11 +110,11 @@ pub struct TimelineInputs {
 ///
 /// Tenant size does not consider the latest state, but only the state until next_gc_cutoff, which
 /// is updated on-demand, during the start of this calculation and separate from the
-/// [`TimelineInputs::latest_gc_cutoff`].
+/// [`Timeline::latest_gc_cutoff`].
 ///
 /// For timelines in general:
 ///
-/// ```text
+/// ```ignore
 /// 0-----|---------|----|------------| · · · · · |·> lsn
 ///   initdb_lsn  branchpoints*  next_gc_cutoff  latest
 /// ```

pageserver/src/tenant/span.rs

@@ -1,17 +0,0 @@
-#[cfg(debug_assertions)]
-use utils::tracing_span_assert::{check_fields_present, MultiNameExtractor};
-
-#[cfg(not(debug_assertions))]
-pub(crate) fn debug_assert_current_span_has_tenant_id() {}
-
-#[cfg(debug_assertions)]
-pub(crate) static TENANT_ID_EXTRACTOR: once_cell::sync::Lazy<MultiNameExtractor<1>> =
-    once_cell::sync::Lazy::new(|| MultiNameExtractor::new("TenantId", ["tenant_id"]));
-
-#[cfg(debug_assertions)]
-#[track_caller]
-pub(crate) fn debug_assert_current_span_has_tenant_id() {
-    if let Err(missing) = check_fields_present!([&*TENANT_ID_EXTRACTOR]) {
-        panic!("missing extractors: {missing:?}")
-    }
-}

pageserver/src/tenant/storage_layer.rs

@@ -41,7 +41,7 @@ pub use inmemory_layer::InMemoryLayer;
 pub use layer_desc::{PersistentLayerDesc, PersistentLayerKey};
 pub use remote_layer::RemoteLayer;
 
-use super::timeline::layer_manager::LayerManager;
+use super::layer_map::BatchedUpdates;
 
 pub fn range_overlaps<T>(a: &Range<T>, b: &Range<T>) -> bool
 where
@@ -54,6 +54,13 @@ where
     }
 }
 
+pub fn range_eq<T>(a: &Range<T>, b: &Range<T>) -> bool
+where
+    T: PartialEq<T>,
+{
+    a.start == b.start && a.end == b.end
+}
+
 /// Struct used to communicate across calls to 'get_value_reconstruct_data'.
 ///
 /// Before first call, you can fill in 'page_img' if you have an older cached
@@ -162,9 +169,6 @@ impl LayerAccessStats {
     /// The caller is responsible for recording a residence event
     /// using [`record_residence_event`] before calling `latest_activity`.
     /// If they don't, [`latest_activity`] will return `None`.
-    ///
-    /// [`record_residence_event`]: Self::record_residence_event
-    /// [`latest_activity`]: Self::latest_activity
     pub(crate) fn empty_will_record_residence_event_later() -> Self {
         LayerAccessStats(Mutex::default())
     }
@@ -172,11 +176,8 @@ impl LayerAccessStats {
     /// Create an empty stats object and record a [`LayerLoad`] event with the given residence status.
     ///
     /// See [`record_residence_event`] for why you need to do this while holding the layer map lock.
-    ///
-    /// [`LayerLoad`]: LayerResidenceEventReason::LayerLoad
-    /// [`record_residence_event`]: Self::record_residence_event
     pub(crate) fn for_loading_layer(
-        layer_map_lock_held_witness: &LayerManager,
+        layer_map_lock_held_witness: &BatchedUpdates<'_>,
         status: LayerResidenceStatus,
     ) -> Self {
         let new = LayerAccessStats(Mutex::new(LayerAccessStatsLocked::default()));
@@ -193,11 +194,9 @@ impl LayerAccessStats {
     /// The `new_status` is not recorded in `self`.
     ///
     /// See [`record_residence_event`] for why you need to do this while holding the layer map lock.
-    ///
-    /// [`record_residence_event`]: Self::record_residence_event
     pub(crate) fn clone_for_residence_change(
         &self,
-        layer_map_lock_held_witness: &LayerManager,
+        layer_map_lock_held_witness: &BatchedUpdates<'_>,
         new_status: LayerResidenceStatus,
     ) -> LayerAccessStats {
         let clone = {
@@ -229,7 +228,7 @@ impl LayerAccessStats {
     ///
     pub(crate) fn record_residence_event(
         &self,
-        _layer_map_lock_held_witness: &LayerManager,
+        _layer_map_lock_held_witness: &BatchedUpdates<'_>,
         status: LayerResidenceStatus,
         reason: LayerResidenceEventReason,
     ) {
@@ -302,13 +301,11 @@ impl LayerAccessStats {
     /// implementation error. This function logs a rate-limited warning in that case.
     ///
     /// TODO: use type system to avoid the need for `fallback`.
-    /// The approach in <https://github.com/neondatabase/neon/pull/3775>
+    /// The approach in https://github.com/neondatabase/neon/pull/3775
     /// could be used to enforce that a residence event is recorded
     /// before a layer is added to the layer map. We could also have
     /// a layer wrapper type that holds the LayerAccessStats, and ensure
     /// that that type can only be produced by inserting into the layer map.
-    ///
-    /// [`record_residence_event`]: Self::record_residence_event
     pub(crate) fn latest_activity(&self) -> Option<SystemTime> {
         let locked = self.0.lock().unwrap();
         let inner = &locked.for_eviction_policy;
@@ -333,7 +330,7 @@ impl LayerAccessStats {
 }
 
 /// Supertrait of the [`Layer`] trait that captures the bare minimum interface
-/// required by [`LayerMap`](super::layer_map::LayerMap).
+/// required by [`LayerMap`].
 ///
 /// All layers should implement a minimal `std::fmt::Debug` without tenant or
 /// timeline names, because those are known in the context of which the layers
@@ -380,18 +377,12 @@ pub trait Layer: std::fmt::Debug + std::fmt::Display + Send + Sync {
     fn dump(&self, verbose: bool, ctx: &RequestContext) -> Result<()>;
 }
 
-/// Returned by [`PersistentLayer::iter`]
+/// Returned by [`Layer::iter`]
 pub type LayerIter<'i> = Box<dyn Iterator<Item = Result<(Key, Lsn, Value)>> + 'i + Send>;
 
-/// Returned by [`PersistentLayer::key_iter`]
+/// Returned by [`Layer::key_iter`]
 pub type LayerKeyIter<'i> = Box<dyn Iterator<Item = (Key, Lsn, u64)> + 'i + Send>;
 
-/// Get a layer descriptor from a layer.
-pub trait AsLayerDesc {
-    /// Get the layer descriptor.
-    fn layer_desc(&self) -> &PersistentLayerDesc;
-}
-
 /// A Layer contains all data in a "rectangle" consisting of a range of keys and
 /// range of LSNs.
 ///
@@ -405,8 +396,10 @@ pub trait AsLayerDesc {
 /// A delta layer contains all modifications within a range of LSNs and keys.
 /// An image layer is a snapshot of all the data in a key-range, at a single
 /// LSN.
-pub trait PersistentLayer: Layer + AsLayerDesc {
-    /// Identify the tenant this layer belongs to
+pub trait PersistentLayer: Layer {
+    /// Get the layer descriptor.
+    fn layer_desc(&self) -> &PersistentLayerDesc;
+
     fn get_tenant_id(&self) -> TenantId {
         self.layer_desc().tenant_id
     }
@@ -472,32 +465,119 @@ pub fn downcast_remote_layer(
 pub mod tests {
     use super::*;
 
-    impl From<DeltaFileName> for PersistentLayerDesc {
+    /// Holds metadata about a layer without any content. Used mostly for testing.
+    ///
+    /// To use filenames as fixtures, parse them as [`LayerFileName`] then convert from that to a
+    /// LayerDescriptor.
+    #[derive(Clone, Debug)]
+    pub struct LayerDescriptor {
+        base: PersistentLayerDesc,
+    }
+
+    impl From<PersistentLayerDesc> for LayerDescriptor {
+        fn from(base: PersistentLayerDesc) -> Self {
+            Self { base }
+        }
+    }
+
+    impl Layer for LayerDescriptor {
+        fn get_value_reconstruct_data(
+            &self,
+            _key: Key,
+            _lsn_range: Range<Lsn>,
+            _reconstruct_data: &mut ValueReconstructState,
+            _ctx: &RequestContext,
+        ) -> Result<ValueReconstructResult> {
+            todo!("This method shouldn't be part of the Layer trait")
+        }
+
+        fn dump(&self, _verbose: bool, _ctx: &RequestContext) -> Result<()> {
+            todo!()
+        }
+
+        /// Boilerplate to implement the Layer trait, always use layer_desc for persistent layers.
+        fn get_key_range(&self) -> Range<Key> {
+            self.layer_desc().key_range.clone()
+        }
+
+        /// Boilerplate to implement the Layer trait, always use layer_desc for persistent layers.
+        fn get_lsn_range(&self) -> Range<Lsn> {
+            self.layer_desc().lsn_range.clone()
+        }
+
+        /// Boilerplate to implement the Layer trait, always use layer_desc for persistent layers.
+        fn is_incremental(&self) -> bool {
+            self.layer_desc().is_incremental
+        }
+    }
+
+    /// Boilerplate to implement the Layer trait, always use layer_desc for persistent layers.
+    impl std::fmt::Display for LayerDescriptor {
+        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+            write!(f, "{}", self.layer_desc().short_id())
+        }
+    }
+
+    impl PersistentLayer for LayerDescriptor {
+        fn layer_desc(&self) -> &PersistentLayerDesc {
+            &self.base
+        }
+
+        fn local_path(&self) -> Option<PathBuf> {
+            unimplemented!()
+        }
+
+        fn iter(&self, _: &RequestContext) -> Result<LayerIter<'_>> {
+            unimplemented!()
+        }
+
+        fn key_iter(&self, _: &RequestContext) -> Result<LayerKeyIter<'_>> {
+            unimplemented!()
+        }
+
+        fn delete_resident_layer_file(&self) -> Result<()> {
+            unimplemented!()
+        }
+
+        fn info(&self, _: LayerAccessStatsReset) -> HistoricLayerInfo {
+            unimplemented!()
+        }
+
+        fn access_stats(&self) -> &LayerAccessStats {
+            unimplemented!()
+        }
+    }
+
+    impl From<DeltaFileName> for LayerDescriptor {
         fn from(value: DeltaFileName) -> Self {
-            PersistentLayerDesc::new_delta(
-                TenantId::from_array([0; 16]),
-                TimelineId::from_array([0; 16]),
-                value.key_range,
-                value.lsn_range,
-                233,
-            )
+            LayerDescriptor {
+                base: PersistentLayerDesc::new_delta(
+                    TenantId::from_array([0; 16]),
+                    TimelineId::from_array([0; 16]),
+                    value.key_range,
+                    value.lsn_range,
+                    233,
+                ),
+            }
         }
     }
 
-    impl From<ImageFileName> for PersistentLayerDesc {
+    impl From<ImageFileName> for LayerDescriptor {
         fn from(value: ImageFileName) -> Self {
-            PersistentLayerDesc::new_img(
-                TenantId::from_array([0; 16]),
-                TimelineId::from_array([0; 16]),
-                value.key_range,
-                value.lsn,
-                false,
-                233,
-            )
+            LayerDescriptor {
+                base: PersistentLayerDesc::new_img(
+                    TenantId::from_array([0; 16]),
+                    TimelineId::from_array([0; 16]),
+                    value.key_range,
+                    value.lsn,
+                    false,
+                    233,
+                ),
+            }
         }
     }
 
-    impl From<LayerFileName> for PersistentLayerDesc {
+    impl From<LayerFileName> for LayerDescriptor {
         fn from(value: LayerFileName) -> Self {
             match value {
                 LayerFileName::Delta(d) => Self::from(d),

pageserver/src/tenant/storage_layer/delta_layer.rs

@@ -7,18 +7,14 @@
 //! must be page images or WAL records with the 'will_init' flag set, so that
 //! they can be replayed without referring to an older page version.
 //!
-//! The delta files are stored in `timelines/<timeline_id>` directory.  Currently,
+//! The delta files are stored in timelines/<timeline_id> directory.  Currently,
 //! there are no subdirectories, and each delta file is named like this:
 //!
-//! ```text
-//!    <key start>-<key end>__<start LSN>-<end LSN>
-//! ```
+//!    <key start>-<key end>__<start LSN>-<end LSN
 //!
 //! For example:
 //!
-//! ```text
 //!    000000067F000032BE0000400000000020B6-000000067F000032BE0000400000000030B6__000000578C6B29-0000000057A50051
-//! ```
 //!
 //! Every delta file consists of three parts: "summary", "index", and
 //! "values". The summary is a fixed size header at the beginning of the file,
@@ -60,8 +56,8 @@ use utils::{
 };
 
 use super::{
-    AsLayerDesc, DeltaFileName, Layer, LayerAccessStats, LayerAccessStatsReset, LayerIter,
-    LayerKeyIter, PathOrConf, PersistentLayerDesc,
+    DeltaFileName, Layer, LayerAccessStats, LayerAccessStatsReset, LayerIter, LayerKeyIter,
+    PathOrConf, PersistentLayerDesc,
 };
 
 ///
@@ -226,14 +222,13 @@ impl Layer for DeltaLayer {
     /// debugging function to print out the contents of the layer
     fn dump(&self, verbose: bool, ctx: &RequestContext) -> Result<()> {
         println!(
-            "----- delta layer for ten {} tli {} keys {}-{} lsn {}-{} size {} ----",
+            "----- delta layer for ten {} tli {} keys {}-{} lsn {}-{} ----",
             self.desc.tenant_id,
             self.desc.timeline_id,
             self.desc.key_range.start,
             self.desc.key_range.end,
             self.desc.lsn_range.start,
-            self.desc.lsn_range.end,
-            self.desc.file_size,
+            self.desc.lsn_range.end
         );
 
         if !verbose {
@@ -407,13 +402,11 @@ impl std::fmt::Display for DeltaLayer {
     }
 }
 
-impl AsLayerDesc for DeltaLayer {
+impl PersistentLayer for DeltaLayer {
     fn layer_desc(&self) -> &PersistentLayerDesc {
         &self.desc
     }
-}
 
-impl PersistentLayer for DeltaLayer {
     fn local_path(&self) -> Option<PathBuf> {
         Some(self.path())
     }
@@ -465,22 +458,22 @@ impl PersistentLayer for DeltaLayer {
 impl DeltaLayer {
     fn path_for(
         path_or_conf: &PathOrConf,
-        tenant_id: &TenantId,
-        timeline_id: &TimelineId,
+        timeline_id: TimelineId,
+        tenant_id: TenantId,
         fname: &DeltaFileName,
     ) -> PathBuf {
         match path_or_conf {
             PathOrConf::Path(path) => path.clone(),
             PathOrConf::Conf(conf) => conf
-                .timeline_path(tenant_id, timeline_id)
+                .timeline_path(&timeline_id, &tenant_id)
                 .join(fname.to_string()),
         }
     }
 
     fn temp_path_for(
         conf: &PageServerConf,
-        tenant_id: &TenantId,
-        timeline_id: &TimelineId,
+        timeline_id: TimelineId,
+        tenant_id: TenantId,
         key_start: Key,
         lsn_range: &Range<Lsn>,
     ) -> PathBuf {
@@ -490,7 +483,7 @@ impl DeltaLayer {
             .map(char::from)
             .collect();
 
-        conf.timeline_path(tenant_id, timeline_id).join(format!(
+        conf.timeline_path(&timeline_id, &tenant_id).join(format!(
             "{}-XXX__{:016X}-{:016X}.{}.{}",
             key_start,
             u64::from(lsn_range.start),
@@ -612,8 +605,8 @@ impl DeltaLayer {
     pub fn path(&self) -> PathBuf {
         Self::path_for(
             &self.path_or_conf,
-            &self.desc.tenant_id,
-            &self.desc.timeline_id,
+            self.desc.timeline_id,
+            self.desc.tenant_id,
             &self.layer_name(),
         )
     }
@@ -661,7 +654,7 @@ impl DeltaLayerWriterInner {
         //
         // Note: This overwrites any existing file. There shouldn't be any.
         // FIXME: throw an error instead?
-        let path = DeltaLayer::temp_path_for(conf, &tenant_id, &timeline_id, key_start, &lsn_range);
+        let path = DeltaLayer::temp_path_for(conf, timeline_id, tenant_id, key_start, &lsn_range);
 
         let mut file = VirtualFile::create(&path)?;
         // make room for the header block
@@ -776,8 +769,8 @@ impl DeltaLayerWriterInner {
         // FIXME: throw an error instead?
         let final_path = DeltaLayer::path_for(
             &PathOrConf::Conf(self.conf),
-            &self.tenant_id,
-            &self.timeline_id,
+            self.timeline_id,
+            self.tenant_id,
             &DeltaFileName {
                 key_range: self.key_start..key_end,
                 lsn_range: self.lsn_range,
@@ -804,7 +797,7 @@ impl DeltaLayerWriterInner {
 ///
 /// # Note
 ///
-/// As described in <https://github.com/neondatabase/neon/issues/2650>, it's
+/// As described in https://github.com/neondatabase/neon/issues/2650, it's
 /// possible for the writer to drop before `finish` is actually called. So this
 /// could lead to odd temporary files in the directory, exhausting file system.
 /// This structure wraps `DeltaLayerWriterInner` and also contains `Drop`

pageserver/src/tenant/storage_layer/filename.rs

@@ -57,9 +57,8 @@ impl Ord for DeltaFileName {
 
 /// Represents the filename of a DeltaLayer
 ///
-/// ```text
 ///    <key start>-<key end>__<LSN start>-<LSN end>
-/// ```
+///
 impl DeltaFileName {
     ///
     /// Parse a string as a delta file name. Returns None if the filename does not
@@ -163,9 +162,7 @@ impl ImageFileName {
 ///
 /// Represents the filename of an ImageLayer
 ///
-/// ```text
 ///    <key start>-<key end>__<LSN>
-/// ```
 impl ImageFileName {
     ///
     /// Parse a string as an image file name. Returns None if the filename does not

pageserver/src/tenant/storage_layer/image_layer.rs

@@ -7,15 +7,11 @@
 //! timelines/<timeline_id> directory.  Currently, there are no
 //! subdirectories, and each image layer file is named like this:
 //!
-//! ```text
 //!    <key start>-<key end>__<LSN>
-//! ```
 //!
 //! For example:
 //!
-//! ```text
 //!    000000067F000032BE0000400000000070B6-000000067F000032BE0000400000000080B6__00000000346BC568
-//! ```
 //!
 //! Every image layer file consists of three parts: "summary",
 //! "index", and "values".  The summary is a fixed size header at the
@@ -57,9 +53,7 @@ use utils::{
 };
 
 use super::filename::ImageFileName;
-use super::{
-    AsLayerDesc, Layer, LayerAccessStatsReset, LayerIter, PathOrConf, PersistentLayerDesc,
-};
+use super::{Layer, LayerAccessStatsReset, LayerIter, PathOrConf, PersistentLayerDesc};
 
 ///
 /// Header stored in the beginning of the file
@@ -159,14 +153,12 @@ impl Layer for ImageLayer {
     /// debugging function to print out the contents of the layer
     fn dump(&self, verbose: bool, ctx: &RequestContext) -> Result<()> {
         println!(
-            "----- image layer for ten {} tli {} key {}-{} at {} is_incremental {} size {} ----",
+            "----- image layer for ten {} tli {} key {}-{} at {} ----",
             self.desc.tenant_id,
             self.desc.timeline_id,
             self.desc.key_range.start,
             self.desc.key_range.end,
-            self.lsn,
-            self.desc.is_incremental,
-            self.desc.file_size
+            self.lsn
         );
 
         if !verbose {
@@ -247,13 +239,11 @@ impl std::fmt::Display for ImageLayer {
     }
 }
 
-impl AsLayerDesc for ImageLayer {
+impl PersistentLayer for ImageLayer {
     fn layer_desc(&self) -> &PersistentLayerDesc {
         &self.desc
     }
-}
 
-impl PersistentLayer for ImageLayer {
     fn local_path(&self) -> Option<PathBuf> {
         Some(self.path())
     }
@@ -296,7 +286,7 @@ impl ImageLayer {
         match path_or_conf {
             PathOrConf::Path(path) => path.to_path_buf(),
             PathOrConf::Conf(conf) => conf
-                .timeline_path(&tenant_id, &timeline_id)
+                .timeline_path(&timeline_id, &tenant_id)
                 .join(fname.to_string()),
         }
     }
@@ -313,7 +303,7 @@ impl ImageLayer {
             .map(char::from)
             .collect();
 
-        conf.timeline_path(&tenant_id, &timeline_id)
+        conf.timeline_path(&timeline_id, &tenant_id)
             .join(format!("{fname}.{rand_string}.{TEMP_FILE_SUFFIX}"))
     }
 
@@ -664,7 +654,7 @@ impl ImageLayerWriterInner {
 ///
 /// # Note
 ///
-/// As described in <https://github.com/neondatabase/neon/issues/2650>, it's
+/// As described in https://github.com/neondatabase/neon/issues/2650, it's
 /// possible for the writer to drop before `finish` is actually called. So this
 /// could lead to odd temporary files in the directory, exhausting file system.
 /// This structure wraps `ImageLayerWriterInner` and also contains `Drop`

pageserver/src/tenant/storage_layer/layer_desc.rs

@@ -174,16 +174,13 @@ impl PersistentLayerDesc {
 
     pub fn dump(&self, _verbose: bool, _ctx: &RequestContext) -> Result<()> {
         println!(
-            "----- layer for ten {} tli {} keys {}-{} lsn {}-{} is_delta {} is_incremental {} size {} ----",
+            "----- layer for ten {} tli {} keys {}-{} lsn {}-{} ----",
             self.tenant_id,
             self.timeline_id,
             self.key_range.start,
             self.key_range.end,
             self.lsn_range.start,
-            self.lsn_range.end,
-            self.is_delta,
-            self.is_incremental,
-            self.file_size,
+            self.lsn_range.end
         );
 
         Ok(())

pageserver/src/tenant/storage_layer/remote_layer.rs

@@ -4,9 +4,9 @@
 use crate::config::PageServerConf;
 use crate::context::RequestContext;
 use crate::repository::Key;
+use crate::tenant::layer_map::BatchedUpdates;
 use crate::tenant::remote_timeline_client::index::LayerFileMetadata;
 use crate::tenant::storage_layer::{Layer, ValueReconstructResult, ValueReconstructState};
-use crate::tenant::timeline::layer_manager::LayerManager;
 use anyhow::{bail, Result};
 use pageserver_api::models::HistoricLayerInfo;
 use std::ops::Range;
@@ -20,12 +20,12 @@ use utils::{
 
 use super::filename::{DeltaFileName, ImageFileName};
 use super::{
-    AsLayerDesc, DeltaLayer, ImageLayer, LayerAccessStats, LayerAccessStatsReset, LayerIter,
-    LayerKeyIter, LayerResidenceStatus, PersistentLayer, PersistentLayerDesc,
+    DeltaLayer, ImageLayer, LayerAccessStats, LayerAccessStatsReset, LayerIter, LayerKeyIter,
+    LayerResidenceStatus, PersistentLayer, PersistentLayerDesc,
 };
 
 /// RemoteLayer is a not yet downloaded [`ImageLayer`] or
-/// [`DeltaLayer`](super::DeltaLayer).
+/// [`crate::storage_layer::DeltaLayer`].
 ///
 /// RemoteLayer might be downloaded on-demand during operations which are
 /// allowed download remote layers and during which, it gets replaced with a
@@ -50,8 +50,6 @@ pub struct RemoteLayer {
     /// It is very unlikely to accumulate these in the Timeline's LayerMap, but having this avoids
     /// a possible fast loop between `Timeline::get_reconstruct_data` and
     /// `Timeline::download_remote_layer`, which also logs.
-    ///
-    /// [`ongoing_download`]: Self::ongoing_download
     pub(crate) download_replacement_failure: std::sync::atomic::AtomicBool,
 }
 
@@ -79,16 +77,13 @@ impl Layer for RemoteLayer {
     /// debugging function to print out the contents of the layer
     fn dump(&self, _verbose: bool, _ctx: &RequestContext) -> Result<()> {
         println!(
-            "----- remote layer for ten {} tli {} keys {}-{} lsn {}-{} is_delta {} is_incremental {} size {} ----",
+            "----- remote layer for ten {} tli {} keys {}-{} lsn {}-{} ----",
             self.desc.tenant_id,
             self.desc.timeline_id,
             self.desc.key_range.start,
             self.desc.key_range.end,
             self.desc.lsn_range.start,
-            self.desc.lsn_range.end,
-            self.desc.is_delta,
-            self.desc.is_incremental,
-            self.desc.file_size,
+            self.desc.lsn_range.end
         );
 
         Ok(())
@@ -117,13 +112,11 @@ impl std::fmt::Display for RemoteLayer {
     }
 }
 
-impl AsLayerDesc for RemoteLayer {
+impl PersistentLayer for RemoteLayer {
     fn layer_desc(&self) -> &PersistentLayerDesc {
         &self.desc
     }
-}
 
-impl PersistentLayer for RemoteLayer {
     fn local_path(&self) -> Option<PathBuf> {
         None
     }
@@ -226,7 +219,7 @@ impl RemoteLayer {
     /// Create a Layer struct representing this layer, after it has been downloaded.
     pub fn create_downloaded_layer(
         &self,
-        layer_map_lock_held_witness: &LayerManager,
+        layer_map_lock_held_witness: &BatchedUpdates<'_>,
         conf: &'static PageServerConf,
         file_size: u64,
     ) -> Arc<dyn PersistentLayer> {

pageserver/src/tenant/timeline/eviction_task.rs

@@ -198,10 +198,10 @@ impl Timeline {
         // So, we just need to deal with this.
         let candidates: Vec<Arc<dyn PersistentLayer>> = {
             let guard = self.layers.read().await;
-            let layers = guard.layer_map();
+            let (layers, mapping) = &*guard;
             let mut candidates = Vec::new();
             for hist_layer in layers.iter_historic_layers() {
-                let hist_layer = guard.get_from_desc(&hist_layer);
+                let hist_layer = mapping.get_from_desc(&hist_layer);
                 if hist_layer.is_remote_layer() {
                     continue;
                 }

pageserver/src/tenant/timeline/layer_manager.rs

@@ -1,370 +0,0 @@
-use anyhow::{bail, ensure, Context, Result};
-use std::{collections::HashMap, sync::Arc};
-use tracing::trace;
-use utils::{
-    id::{TenantId, TimelineId},
-    lsn::{AtomicLsn, Lsn},
-};
-
-use crate::{
-    config::PageServerConf,
-    metrics::TimelineMetrics,
-    tenant::{
-        layer_map::{BatchedUpdates, LayerMap},
-        storage_layer::{
-            AsLayerDesc, DeltaLayer, ImageLayer, InMemoryLayer, Layer, PersistentLayer,
-            PersistentLayerDesc, PersistentLayerKey, RemoteLayer,
-        },
-        timeline::compare_arced_layers,
-    },
-};
-
-/// Provides semantic APIs to manipulate the layer map.
-pub struct LayerManager {
-    layer_map: LayerMap,
-    layer_fmgr: LayerFileManager,
-}
-
-/// After GC, the layer map changes will not be applied immediately. Users should manually apply the changes after
-/// scheduling deletes in remote client.
-pub struct ApplyGcResultGuard<'a>(BatchedUpdates<'a>);
-
-impl ApplyGcResultGuard<'_> {
-    pub fn flush(self) {
-        self.0.flush();
-    }
-}
-
-impl LayerManager {
-    pub fn create() -> Self {
-        Self {
-            layer_map: LayerMap::default(),
-            layer_fmgr: LayerFileManager::new(),
-        }
-    }
-
-    pub fn get_from_desc(&self, desc: &PersistentLayerDesc) -> Arc<dyn PersistentLayer> {
-        self.layer_fmgr.get_from_desc(desc)
-    }
-
-    /// Get an immutable reference to the layer map.
-    ///
-    /// We expect users only to be able to get an immutable layer map. If users want to make modifications,
-    /// they should use the below semantic APIs. This design makes us step closer to immutable storage state.
-    pub fn layer_map(&self) -> &LayerMap {
-        &self.layer_map
-    }
-
-    /// Get a mutable reference to the layer map. This function will be removed once `flush_frozen_layer`
-    /// gets a refactor.
-    pub fn layer_map_mut(&mut self) -> &mut LayerMap {
-        &mut self.layer_map
-    }
-
-    /// Replace layers in the layer file manager, used in evictions and layer downloads.
-    pub fn replace_and_verify(
-        &mut self,
-        expected: Arc<dyn PersistentLayer>,
-        new: Arc<dyn PersistentLayer>,
-    ) -> Result<()> {
-        self.layer_fmgr.replace_and_verify(expected, new)
-    }
-
-    /// Called from `load_layer_map`. Initialize the layer manager with:
-    /// 1. all on-disk layers
-    /// 2. next open layer (with disk disk_consistent_lsn LSN)
-    pub fn initialize_local_layers(
-        &mut self,
-        on_disk_layers: Vec<Arc<dyn PersistentLayer>>,
-        next_open_layer_at: Lsn,
-    ) {
-        let mut updates = self.layer_map.batch_update();
-        for layer in on_disk_layers {
-            Self::insert_historic_layer(layer, &mut updates, &mut self.layer_fmgr);
-        }
-        updates.flush();
-        self.layer_map.next_open_layer_at = Some(next_open_layer_at);
-    }
-
-    /// Initialize when creating a new timeline, called in `init_empty_layer_map`.
-    pub fn initialize_empty(&mut self, next_open_layer_at: Lsn) {
-        self.layer_map.next_open_layer_at = Some(next_open_layer_at);
-    }
-
-    pub fn initialize_remote_layers(
-        &mut self,
-        corrupted_local_layers: Vec<Arc<dyn PersistentLayer>>,
-        remote_layers: Vec<Arc<RemoteLayer>>,
-    ) {
-        let mut updates = self.layer_map.batch_update();
-        for layer in corrupted_local_layers {
-            Self::remove_historic_layer(layer, &mut updates, &mut self.layer_fmgr);
-        }
-        for layer in remote_layers {
-            Self::insert_historic_layer(layer, &mut updates, &mut self.layer_fmgr);
-        }
-        updates.flush();
-    }
-
-    /// Open a new writable layer to append data if there is no open layer, otherwise return the current open layer,
-    /// called within `get_layer_for_write`.
-    pub fn get_layer_for_write(
-        &mut self,
-        lsn: Lsn,
-        last_record_lsn: Lsn,
-        conf: &'static PageServerConf,
-        timeline_id: TimelineId,
-        tenant_id: TenantId,
-    ) -> Result<Arc<InMemoryLayer>> {
-        ensure!(lsn.is_aligned());
-
-        ensure!(
-            lsn > last_record_lsn,
-            "cannot modify relation after advancing last_record_lsn (incoming_lsn={}, last_record_lsn={})\n{}",
-            lsn,
-            last_record_lsn,
-            std::backtrace::Backtrace::force_capture(),
-        );
-
-        // Do we have a layer open for writing already?
-        let layer = if let Some(open_layer) = &self.layer_map.open_layer {
-            if open_layer.get_lsn_range().start > lsn {
-                bail!(
-                    "unexpected open layer in the future: open layers starts at {}, write lsn {}",
-                    open_layer.get_lsn_range().start,
-                    lsn
-                );
-            }
-
-            Arc::clone(open_layer)
-        } else {
-            // No writeable layer yet. Create one.
-            let start_lsn = self
-                .layer_map
-                .next_open_layer_at
-                .context("No next open layer found")?;
-
-            trace!(
-                "creating in-memory layer at {}/{} for record at {}",
-                timeline_id,
-                start_lsn,
-                lsn
-            );
-
-            let new_layer = InMemoryLayer::create(conf, timeline_id, tenant_id, start_lsn)?;
-            let layer = Arc::new(new_layer);
-
-            self.layer_map.open_layer = Some(layer.clone());
-            self.layer_map.next_open_layer_at = None;
-
-            layer
-        };
-
-        Ok(layer)
-    }
-
-    /// Called from `freeze_inmem_layer`, returns true if successfully frozen.
-    pub fn try_freeze_in_memory_layer(
-        &mut self,
-        Lsn(last_record_lsn): Lsn,
-        last_freeze_at: &AtomicLsn,
-    ) {
-        let end_lsn = Lsn(last_record_lsn + 1);
-
-        if let Some(open_layer) = &self.layer_map.open_layer {
-            let open_layer_rc = Arc::clone(open_layer);
-            // Does this layer need freezing?
-            open_layer.freeze(end_lsn);
-
-            // The layer is no longer open, update the layer map to reflect this.
-            // We will replace it with on-disk historics below.
-            self.layer_map.frozen_layers.push_back(open_layer_rc);
-            self.layer_map.open_layer = None;
-            self.layer_map.next_open_layer_at = Some(end_lsn);
-            last_freeze_at.store(end_lsn);
-        }
-    }
-
-    /// Add image layers to the layer map, called from `create_image_layers`.
-    pub fn track_new_image_layers(&mut self, image_layers: Vec<ImageLayer>) {
-        let mut updates = self.layer_map.batch_update();
-        for layer in image_layers {
-            Self::insert_historic_layer(Arc::new(layer), &mut updates, &mut self.layer_fmgr);
-        }
-        updates.flush();
-    }
-
-    /// Insert into the layer map when a new delta layer is created, called from `create_delta_layer`.
-    pub fn track_new_l0_delta_layer(&mut self, delta_layer: Arc<DeltaLayer>) {
-        let mut updates = self.layer_map.batch_update();
-        Self::insert_historic_layer(delta_layer, &mut updates, &mut self.layer_fmgr);
-        updates.flush();
-    }
-
-    /// Called when compaction is completed.
-    pub fn finish_compact_l0(
-        &mut self,
-        layer_removal_cs: Arc<tokio::sync::OwnedMutexGuard<()>>,
-        compact_from: Vec<Arc<dyn PersistentLayer>>,
-        compact_to: Vec<Arc<dyn PersistentLayer>>,
-        metrics: &TimelineMetrics,
-    ) -> Result<()> {
-        let mut updates = self.layer_map.batch_update();
-        for l in compact_to {
-            Self::insert_historic_layer(l, &mut updates, &mut self.layer_fmgr);
-        }
-        for l in compact_from {
-            // NB: the layer file identified by descriptor `l` is guaranteed to be present
-            // in the LayerFileManager because compaction kept holding `layer_removal_cs` the entire
-            // time, even though we dropped `Timeline::layers` inbetween.
-            Self::delete_historic_layer(
-                layer_removal_cs.clone(),
-                l,
-                &mut updates,
-                metrics,
-                &mut self.layer_fmgr,
-            )?;
-        }
-        updates.flush();
-        Ok(())
-    }
-
-    /// Called when garbage collect the timeline. Returns a guard that will apply the updates to the layer map.
-    pub fn finish_gc_timeline(
-        &mut self,
-        layer_removal_cs: Arc<tokio::sync::OwnedMutexGuard<()>>,
-        gc_layers: Vec<Arc<dyn PersistentLayer>>,
-        metrics: &TimelineMetrics,
-    ) -> Result<ApplyGcResultGuard> {
-        let mut updates = self.layer_map.batch_update();
-        for doomed_layer in gc_layers {
-            Self::delete_historic_layer(
-                layer_removal_cs.clone(),
-                doomed_layer,
-                &mut updates,
-                metrics,
-                &mut self.layer_fmgr,
-            )?; // FIXME: schedule succeeded deletions in timeline.rs `gc_timeline` instead of in batch?
-        }
-        Ok(ApplyGcResultGuard(updates))
-    }
-
-    /// Helper function to insert a layer into the layer map and file manager.
-    fn insert_historic_layer(
-        layer: Arc<dyn PersistentLayer>,
-        updates: &mut BatchedUpdates<'_>,
-        mapping: &mut LayerFileManager,
-    ) {
-        updates.insert_historic(layer.layer_desc().clone());
-        mapping.insert(layer);
-    }
-
-    /// Helper function to remove a layer into the layer map and file manager
-    fn remove_historic_layer(
-        layer: Arc<dyn PersistentLayer>,
-        updates: &mut BatchedUpdates<'_>,
-        mapping: &mut LayerFileManager,
-    ) {
-        updates.remove_historic(layer.layer_desc().clone());
-        mapping.remove(layer);
-    }
-
-    /// Removes the layer from local FS (if present) and from memory.
-    /// Remote storage is not affected by this operation.
-    fn delete_historic_layer(
-        // we cannot remove layers otherwise, since gc and compaction will race
-        _layer_removal_cs: Arc<tokio::sync::OwnedMutexGuard<()>>,
-        layer: Arc<dyn PersistentLayer>,
-        updates: &mut BatchedUpdates<'_>,
-        metrics: &TimelineMetrics,
-        mapping: &mut LayerFileManager,
-    ) -> anyhow::Result<()> {
-        if !layer.is_remote_layer() {
-            layer.delete_resident_layer_file()?;
-            let layer_file_size = layer.file_size();
-            metrics.resident_physical_size_gauge.sub(layer_file_size);
-        }
-
-        // TODO Removing from the bottom of the layer map is expensive.
-        //      Maybe instead discard all layer map historic versions that
-        //      won't be needed for page reconstruction for this timeline,
-        //      and mark what we can't delete yet as deleted from the layer
-        //      map index without actually rebuilding the index.
-        updates.remove_historic(layer.layer_desc().clone());
-        mapping.remove(layer);
-
-        Ok(())
-    }
-}
-
-pub struct LayerFileManager<T: AsLayerDesc + ?Sized = dyn PersistentLayer>(
-    HashMap<PersistentLayerKey, Arc<T>>,
-);
-
-impl<T: AsLayerDesc + ?Sized> LayerFileManager<T> {
-    fn get_from_desc(&self, desc: &PersistentLayerDesc) -> Arc<T> {
-        // The assumption for the `expect()` is that all code maintains the following invariant:
-        // A layer's descriptor is present in the LayerMap => the LayerFileManager contains a layer for the descriptor.
-        self.0
-            .get(&desc.key())
-            .with_context(|| format!("get layer from desc: {}", desc.filename()))
-            .expect("not found")
-            .clone()
-    }
-
-    pub(crate) fn insert(&mut self, layer: Arc<T>) {
-        let present = self.0.insert(layer.layer_desc().key(), layer.clone());
-        if present.is_some() && cfg!(debug_assertions) {
-            panic!("overwriting a layer: {:?}", layer.layer_desc())
-        }
-    }
-
-    pub(crate) fn new() -> Self {
-        Self(HashMap::new())
-    }
-
-    pub(crate) fn remove(&mut self, layer: Arc<T>) {
-        let present = self.0.remove(&layer.layer_desc().key());
-        if present.is_none() && cfg!(debug_assertions) {
-            panic!(
-                "removing layer that is not present in layer mapping: {:?}",
-                layer.layer_desc()
-            )
-        }
-    }
-
-    pub(crate) fn replace_and_verify(&mut self, expected: Arc<T>, new: Arc<T>) -> Result<()> {
-        let key = expected.layer_desc().key();
-        let other = new.layer_desc().key();
-
-        let expected_l0 = LayerMap::is_l0(expected.layer_desc());
-        let new_l0 = LayerMap::is_l0(new.layer_desc());
-
-        fail::fail_point!("layermap-replace-notfound", |_| anyhow::bail!(
-            "layermap-replace-notfound"
-        ));
-
-        anyhow::ensure!(
-            key == other,
-            "expected and new layer have different keys: {key:?} != {other:?}"
-        );
-
-        anyhow::ensure!(
-            expected_l0 == new_l0,
-            "one layer is l0 while the other is not: {expected_l0} != {new_l0}"
-        );
-
-        if let Some(layer) = self.0.get_mut(&key) {
-            anyhow::ensure!(
-                compare_arced_layers(&expected, layer),
-                "another layer was found instead of expected, expected={expected:?}, new={new:?}",
-                expected = Arc::as_ptr(&expected),
-                new = Arc::as_ptr(layer),
-            );
-            *layer = new;
-            Ok(())
-        } else {
-            anyhow::bail!("layer was not found");
-        }
-    }
-}

pageserver/src/tenant/timeline/logical_size.rs

@@ -1,128 +0,0 @@
-use anyhow::Context;
-use once_cell::sync::OnceCell;
-
-use tokio::sync::Semaphore;
-use utils::lsn::Lsn;
-
-use std::sync::atomic::{AtomicI64, Ordering as AtomicOrdering};
-use std::sync::Arc;
-
-/// Internal structure to hold all data needed for logical size calculation.
-///
-/// Calculation consists of two stages:
-///
-/// 1. Initial size calculation. That might take a long time, because it requires
-/// reading all layers containing relation sizes at `initial_part_end`.
-///
-/// 2. Collecting an incremental part and adding that to the initial size.
-/// Increments are appended on walreceiver writing new timeline data,
-/// which result in increase or decrease of the logical size.
-pub(super) struct LogicalSize {
-    /// Size, potentially slow to compute. Calculating this might require reading multiple
-    /// layers, and even ancestor's layers.
-    ///
-    /// NOTE: size at a given LSN is constant, but after a restart we will calculate
-    /// the initial size at a different LSN.
-    pub initial_logical_size: OnceCell<u64>,
-
-    /// Semaphore to track ongoing calculation of `initial_logical_size`.
-    pub initial_size_computation: Arc<tokio::sync::Semaphore>,
-
-    /// Latest Lsn that has its size uncalculated, could be absent for freshly created timelines.
-    pub initial_part_end: Option<Lsn>,
-
-    /// All other size changes after startup, combined together.
-    ///
-    /// Size shouldn't ever be negative, but this is signed for two reasons:
-    ///
-    /// 1. If we initialized the "baseline" size lazily, while we already
-    /// process incoming WAL, the incoming WAL records could decrement the
-    /// variable and temporarily make it negative. (This is just future-proofing;
-    /// the initialization is currently not done lazily.)
-    ///
-    /// 2. If there is a bug and we e.g. forget to increment it in some cases
-    /// when size grows, but remember to decrement it when it shrinks again, the
-    /// variable could go negative. In that case, it seems better to at least
-    /// try to keep tracking it, rather than clamp or overflow it. Note that
-    /// get_current_logical_size() will clamp the returned value to zero if it's
-    /// negative, and log an error. Could set it permanently to zero or some
-    /// special value to indicate "broken" instead, but this will do for now.
-    ///
-    /// Note that we also expose a copy of this value as a prometheus metric,
-    /// see `current_logical_size_gauge`. Use the `update_current_logical_size`
-    /// to modify this, it will also keep the prometheus metric in sync.
-    pub size_added_after_initial: AtomicI64,
-}
-
-/// Normalized current size, that the data in pageserver occupies.
-#[derive(Debug, Clone, Copy)]
-pub(super) enum CurrentLogicalSize {
-    /// The size is not yet calculated to the end, this is an intermediate result,
-    /// constructed from walreceiver increments and normalized: logical data could delete some objects, hence be negative,
-    /// yet total logical size cannot be below 0.
-    Approximate(u64),
-    // Fully calculated logical size, only other future walreceiver increments are changing it, and those changes are
-    // available for observation without any calculations.
-    Exact(u64),
-}
-
-impl CurrentLogicalSize {
-    pub(super) fn size(&self) -> u64 {
-        *match self {
-            Self::Approximate(size) => size,
-            Self::Exact(size) => size,
-        }
-    }
-}
-
-impl LogicalSize {
-    pub(super) fn empty_initial() -> Self {
-        Self {
-            initial_logical_size: OnceCell::with_value(0),
-            //  initial_logical_size already computed, so, don't admit any calculations
-            initial_size_computation: Arc::new(Semaphore::new(0)),
-            initial_part_end: None,
-            size_added_after_initial: AtomicI64::new(0),
-        }
-    }
-
-    pub(super) fn deferred_initial(compute_to: Lsn) -> Self {
-        Self {
-            initial_logical_size: OnceCell::new(),
-            initial_size_computation: Arc::new(Semaphore::new(1)),
-            initial_part_end: Some(compute_to),
-            size_added_after_initial: AtomicI64::new(0),
-        }
-    }
-
-    pub(super) fn current_size(&self) -> anyhow::Result<CurrentLogicalSize> {
-        let size_increment: i64 = self.size_added_after_initial.load(AtomicOrdering::Acquire);
-        //                  ^^^ keep this type explicit so that the casts in this function break if
-        //                  we change the type.
-        match self.initial_logical_size.get() {
-            Some(initial_size) => {
-                initial_size.checked_add_signed(size_increment)
-                    .with_context(|| format!("Overflow during logical size calculation, initial_size: {initial_size}, size_increment: {size_increment}"))
-                    .map(CurrentLogicalSize::Exact)
-            }
-            None => {
-                let non_negative_size_increment = u64::try_from(size_increment).unwrap_or(0);
-                Ok(CurrentLogicalSize::Approximate(non_negative_size_increment))
-            }
-        }
-    }
-
-    pub(super) fn increment_size(&self, delta: i64) {
-        self.size_added_after_initial
-            .fetch_add(delta, AtomicOrdering::SeqCst);
-    }
-
-    /// Make the value computed by initial logical size computation
-    /// available for re-use. This doesn't contain the incremental part.
-    pub(super) fn initialized_size(&self, lsn: Lsn) -> Option<u64> {
-        match self.initial_part_end {
-            Some(v) if v == lsn => self.initial_logical_size.get().copied(),
-            _ => None,
-        }
-    }
-}

pageserver/src/tenant/timeline/span.rs

@@ -1,20 +0,0 @@
-#[cfg(debug_assertions)]
-use utils::tracing_span_assert::{check_fields_present, Extractor, MultiNameExtractor};
-
-#[cfg(not(debug_assertions))]
-pub(crate) fn debug_assert_current_span_has_tenant_and_timeline_id() {}
-
-#[cfg(debug_assertions)]
-#[track_caller]
-pub(crate) fn debug_assert_current_span_has_tenant_and_timeline_id() {
-    static TIMELINE_ID_EXTRACTOR: once_cell::sync::Lazy<MultiNameExtractor<1>> =
-        once_cell::sync::Lazy::new(|| MultiNameExtractor::new("TimelineId", ["timeline_id"]));
-
-    let fields: [&dyn Extractor; 2] = [
-        &*crate::tenant::span::TENANT_ID_EXTRACTOR,
-        &*TIMELINE_ID_EXTRACTOR,
-    ];
-    if let Err(missing) = check_fields_present!(fields) {
-        panic!("missing extractors: {missing:?}")
-    }
-}

pageserver/src/tenant/timeline/uninit.rs

@@ -1,219 +0,0 @@
-use std::{collections::hash_map::Entry, fs, path::PathBuf, sync::Arc};
-
-use anyhow::Context;
-use tracing::{error, info, info_span, warn};
-use utils::{crashsafe, id::TimelineId, lsn::Lsn};
-
-use crate::{
-    context::RequestContext,
-    import_datadir,
-    tenant::{ignore_absent_files, Tenant},
-};
-
-use super::Timeline;
-
-/// A timeline with some of its files on disk, being initialized.
-/// This struct ensures the atomicity of the timeline init: it's either properly created and inserted into pageserver's memory, or
-/// its local files are removed. In the worst case of a crash, an uninit mark file is left behind, which causes the directory
-/// to be removed on next restart.
-///
-/// The caller is responsible for proper timeline data filling before the final init.
-#[must_use]
-pub struct UninitializedTimeline<'t> {
-    pub(crate) owning_tenant: &'t Tenant,
-    timeline_id: TimelineId,
-    raw_timeline: Option<(Arc<Timeline>, TimelineUninitMark)>,
-}
-
-impl<'t> UninitializedTimeline<'t> {
-    pub(crate) fn new(
-        owning_tenant: &'t Tenant,
-        timeline_id: TimelineId,
-        raw_timeline: Option<(Arc<Timeline>, TimelineUninitMark)>,
-    ) -> Self {
-        Self {
-            owning_tenant,
-            timeline_id,
-            raw_timeline,
-        }
-    }
-
-    /// Finish timeline creation: insert it into the Tenant's timelines map and remove the
-    /// uninit mark file.
-    ///
-    /// This function launches the flush loop if not already done.
-    ///
-    /// The caller is responsible for activating the timeline (function `.activate()`).
-    pub(crate) fn finish_creation(mut self) -> anyhow::Result<Arc<Timeline>> {
-        let timeline_id = self.timeline_id;
-        let tenant_id = self.owning_tenant.tenant_id;
-
-        let (new_timeline, uninit_mark) = self.raw_timeline.take().with_context(|| {
-            format!("No timeline for initalization found for {tenant_id}/{timeline_id}")
-        })?;
-
-        // Check that the caller initialized disk_consistent_lsn
-        let new_disk_consistent_lsn = new_timeline.get_disk_consistent_lsn();
-        anyhow::ensure!(
-            new_disk_consistent_lsn.is_valid(),
-            "new timeline {tenant_id}/{timeline_id} has invalid disk_consistent_lsn"
-        );
-
-        let mut timelines = self.owning_tenant.timelines.lock().unwrap();
-        match timelines.entry(timeline_id) {
-            Entry::Occupied(_) => anyhow::bail!(
-                "Found freshly initialized timeline {tenant_id}/{timeline_id} in the tenant map"
-            ),
-            Entry::Vacant(v) => {
-                uninit_mark.remove_uninit_mark().with_context(|| {
-                    format!(
-                        "Failed to remove uninit mark file for timeline {tenant_id}/{timeline_id}"
-                    )
-                })?;
-                v.insert(Arc::clone(&new_timeline));
-
-                new_timeline.maybe_spawn_flush_loop();
-            }
-        }
-
-        Ok(new_timeline)
-    }
-
-    /// Prepares timeline data by loading it from the basebackup archive.
-    pub(crate) async fn import_basebackup_from_tar(
-        self,
-        copyin_read: &mut (impl tokio::io::AsyncRead + Send + Sync + Unpin),
-        base_lsn: Lsn,
-        broker_client: storage_broker::BrokerClientChannel,
-        ctx: &RequestContext,
-    ) -> anyhow::Result<Arc<Timeline>> {
-        let raw_timeline = self.raw_timeline()?;
-
-        import_datadir::import_basebackup_from_tar(raw_timeline, copyin_read, base_lsn, ctx)
-            .await
-            .context("Failed to import basebackup")?;
-
-        // Flush the new layer files to disk, before we make the timeline as available to
-        // the outside world.
-        //
-        // Flush loop needs to be spawned in order to be able to flush.
-        raw_timeline.maybe_spawn_flush_loop();
-
-        fail::fail_point!("before-checkpoint-new-timeline", |_| {
-            anyhow::bail!("failpoint before-checkpoint-new-timeline");
-        });
-
-        raw_timeline
-            .freeze_and_flush()
-            .await
-            .context("Failed to flush after basebackup import")?;
-
-        // All the data has been imported. Insert the Timeline into the tenant's timelines
-        // map and remove the uninit mark file.
-        let tl = self.finish_creation()?;
-        tl.activate(broker_client, None, ctx);
-        Ok(tl)
-    }
-
-    pub(crate) fn raw_timeline(&self) -> anyhow::Result<&Arc<Timeline>> {
-        Ok(&self
-            .raw_timeline
-            .as_ref()
-            .with_context(|| {
-                format!(
-                    "No raw timeline {}/{} found",
-                    self.owning_tenant.tenant_id, self.timeline_id
-                )
-            })?
-            .0)
-    }
-}
-
-impl Drop for UninitializedTimeline<'_> {
-    fn drop(&mut self) {
-        if let Some((_, uninit_mark)) = self.raw_timeline.take() {
-            let _entered = info_span!("drop_uninitialized_timeline", tenant_id = %self.owning_tenant.tenant_id, timeline_id = %self.timeline_id).entered();
-            error!("Timeline got dropped without initializing, cleaning its files");
-            cleanup_timeline_directory(uninit_mark);
-        }
-    }
-}
-
-pub(crate) fn cleanup_timeline_directory(uninit_mark: TimelineUninitMark) {
-    let timeline_path = &uninit_mark.timeline_path;
-    match ignore_absent_files(|| fs::remove_dir_all(timeline_path)) {
-        Ok(()) => {
-            info!("Timeline dir {timeline_path:?} removed successfully, removing the uninit mark")
-        }
-        Err(e) => {
-            error!("Failed to clean up uninitialized timeline directory {timeline_path:?}: {e:?}")
-        }
-    }
-    drop(uninit_mark); // mark handles its deletion on drop, gets retained if timeline dir exists
-}
-
-/// An uninit mark file, created along the timeline dir to ensure the timeline either gets fully initialized and loaded into pageserver's memory,
-/// or gets removed eventually.
-///
-/// XXX: it's important to create it near the timeline dir, not inside it to ensure timeline dir gets removed first.
-#[must_use]
-pub(crate) struct TimelineUninitMark {
-    uninit_mark_deleted: bool,
-    uninit_mark_path: PathBuf,
-    pub(crate) timeline_path: PathBuf,
-}
-
-impl TimelineUninitMark {
-    pub(crate) fn new(uninit_mark_path: PathBuf, timeline_path: PathBuf) -> Self {
-        Self {
-            uninit_mark_deleted: false,
-            uninit_mark_path,
-            timeline_path,
-        }
-    }
-
-    fn remove_uninit_mark(mut self) -> anyhow::Result<()> {
-        if !self.uninit_mark_deleted {
-            self.delete_mark_file_if_present()?;
-        }
-
-        Ok(())
-    }
-
-    fn delete_mark_file_if_present(&mut self) -> anyhow::Result<()> {
-        let uninit_mark_file = &self.uninit_mark_path;
-        let uninit_mark_parent = uninit_mark_file
-            .parent()
-            .with_context(|| format!("Uninit mark file {uninit_mark_file:?} has no parent"))?;
-        ignore_absent_files(|| fs::remove_file(uninit_mark_file)).with_context(|| {
-            format!("Failed to remove uninit mark file at path {uninit_mark_file:?}")
-        })?;
-        crashsafe::fsync(uninit_mark_parent).context("Failed to fsync uninit mark parent")?;
-        self.uninit_mark_deleted = true;
-
-        Ok(())
-    }
-}
-
-impl Drop for TimelineUninitMark {
-    fn drop(&mut self) {
-        if !self.uninit_mark_deleted {
-            if self.timeline_path.exists() {
-                error!(
-                    "Uninit mark {} is not removed, timeline {} stays uninitialized",
-                    self.uninit_mark_path.display(),
-                    self.timeline_path.display()
-                )
-            } else {
-                // unblock later timeline creation attempts
-                warn!(
-                    "Removing intermediate uninit mark file {}",
-                    self.uninit_mark_path.display()
-                );
-                if let Err(e) = self.delete_mark_file_if_present() {
-                    error!("Failed to remove the uninit mark file: {e}")
-                }
-            }
-        }
-    }
-}

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

@@ -6,7 +6,7 @@
 //! Current connection state is tracked too, to ensure it's not getting stale.
 //!
 //! After every connection or storage broker update fetched, the state gets updated correspondingly and rechecked for the new conneciton leader,
-//! then a (re)connection happens, if necessary.
+//! then a [re]connection happens, if necessary.
 //! Only WAL streaming task expects to be finished, other loops (storage broker, connection management) never exit unless cancelled explicitly via the dedicated channel.
 
 use std::{collections::HashMap, num::NonZeroU64, ops::ControlFlow, sync::Arc, time::Duration};
@@ -266,7 +266,7 @@ pub struct ConnectionManagerStatus {
 impl ConnectionManagerStatus {
     /// Generates a string, describing current connection status in a form, suitable for logging.
     pub fn to_human_readable_string(&self) -> String {
-        let mut resulting_string = String::new();
+        let mut resulting_string = "WalReceiver status".to_string();
         match &self.existing_connection {
             Some(connection) => {
                 if connection.has_processed_wal {

pageserver/src/walrecord.rs

@@ -360,7 +360,6 @@ impl XlXactParsedRecord {
             }
         }
         let mut xnodes = Vec::<RelFileNode>::new();
-        // In v16 this XACT_XINFO_HAS_RELFILENODES is renamed to XACT_XINFO_HAS_RELFILELOCATORS
         if xinfo & pg_constants::XACT_XINFO_HAS_RELFILENODES != 0 {
             let nrels = buf.get_i32_le();
             for _i in 0..nrels {

pageserver/src/walredo.rs

@@ -175,8 +175,8 @@ impl WalRedoManager for PostgresRedoManager {
         let mut img = base_img.map(|p| p.1);
         let mut batch_neon = can_apply_in_neon(&records[0].1);
         let mut batch_start = 0;
-        for (i, record) in records.iter().enumerate().skip(1) {
-            let rec_neon = can_apply_in_neon(&record.1);
+        for i in 1..records.len() {
+            let rec_neon = can_apply_in_neon(&records[i].1);
 
             if rec_neon != batch_neon {
                 let result = if batch_neon {
@@ -685,7 +685,7 @@ impl PostgresRedoManager {
             // as close-on-exec by default, but that's not enough, since we use
             // libraries that directly call libc open without setting that flag.
             .close_fds()
-            .spawn_no_leak_child(self.tenant_id)
+            .spawn_no_leak_child()
             .map_err(|e| {
                 Error::new(
                     e.kind(),
@@ -989,7 +989,6 @@ impl PostgresRedoManager {
 /// Wrapper type around `std::process::Child` which guarantees that the child
 /// will be killed and waited-for by this process before being dropped.
 struct NoLeakChild {
-    tenant_id: TenantId,
     child: Option<Child>,
 }
 
@@ -1008,12 +1007,9 @@ impl DerefMut for NoLeakChild {
 }
 
 impl NoLeakChild {
-    fn spawn(tenant_id: TenantId, command: &mut Command) -> io::Result<Self> {
+    fn spawn(command: &mut Command) -> io::Result<Self> {
         let child = command.spawn()?;
-        Ok(NoLeakChild {
-            tenant_id,
-            child: Some(child),
-        })
+        Ok(NoLeakChild { child: Some(child) })
     }
 
     fn kill_and_wait(mut self) {
@@ -1060,16 +1056,11 @@ impl Drop for NoLeakChild {
             Some(child) => child,
             None => return,
         };
-        let tenant_id = self.tenant_id;
         // Offload the kill+wait of the child process into the background.
         // If someone stops the runtime, we'll leak the child process.
         // We can ignore that case because we only stop the runtime on pageserver exit.
         BACKGROUND_RUNTIME.spawn(async move {
             tokio::task::spawn_blocking(move || {
-                // Intentionally don't inherit the tracing context from whoever is dropping us.
-                // This thread here is going to outlive of our dropper.
-                let span = tracing::info_span!("walredo", %tenant_id);
-                let _entered = span.enter();
                 Self::kill_and_wait_impl(child);
             })
             .await
@@ -1078,12 +1069,12 @@ impl Drop for NoLeakChild {
 }
 
 trait NoLeakChildCommandExt {
-    fn spawn_no_leak_child(&mut self, tenant_id: TenantId) -> io::Result<NoLeakChild>;
+    fn spawn_no_leak_child(&mut self) -> io::Result<NoLeakChild>;
 }
 
 impl NoLeakChildCommandExt for Command {
-    fn spawn_no_leak_child(&mut self, tenant_id: TenantId) -> io::Result<NoLeakChild> {
-        NoLeakChild::spawn(tenant_id, self)
+    fn spawn_no_leak_child(&mut self) -> io::Result<NoLeakChild> {
+        NoLeakChild::spawn(self)
     }
 }
 

pgxn/neon/file_cache.c

@@ -25,11 +25,7 @@
 #include "pagestore_client.h"
 #include "access/parallel.h"
 #include "postmaster/bgworker.h"
-#if PG_VERSION_NUM >= 160000
-#include "storage/relfilelocator.h"
-#else
 #include "storage/relfilenode.h"
-#endif
 #include "storage/buf_internals.h"
 #include "storage/latch.h"
 #include "storage/ipc.h"
@@ -43,7 +39,6 @@
 #include "postmaster/bgworker.h"
 #include "postmaster/interrupt.h"
 
-
 /*
  * Local file cache is used to temporary store relations pages in local file system.
  * All blocks of all relations are stored inside one file and addressed using shared hash map.
@@ -365,12 +360,9 @@ lfc_cache_contains(RelFileNode rnode, ForkNumber forkNum, BlockNumber blkno)
 	if (lfc_size_limit == 0) /* fast exit if file cache is disabled */
 		return false;
 
-#if PG_VERSION_NUM >= 160000
-	InitBufferTag(&tag, &rnode, forkNum, (blkno & ~(BLOCKS_PER_CHUNK-1)));
-#else
-	INIT_BUFFERTAG(tag, rnode, forkNum, (blkno & ~(BLOCKS_PER_CHUNK-1)));
-#endif
-
+	tag.rnode = rnode;
+	tag.forkNum = forkNum;
+	tag.blockNum = blkno & ~(BLOCKS_PER_CHUNK-1);
 	hash = get_hash_value(lfc_hash, &tag);
 
 	LWLockAcquire(lfc_lock, LW_SHARED);
@@ -395,11 +387,7 @@ lfc_evict(RelFileNode rnode, ForkNumber forkNum, BlockNumber blkno)
 	if (lfc_size_limit == 0) /* fast exit if file cache is disabled */
 		return;
 
-#if PG_VERSION_NUM >= 160000
-	InitBufferTag(&tag, &rnode, forkNum, (blkno & ~(BLOCKS_PER_CHUNK-1)));
-#else
 	INIT_BUFFERTAG(tag, rnode, forkNum, (blkno & ~(BLOCKS_PER_CHUNK-1)));
-#endif
 
 	hash = get_hash_value(lfc_hash, &tag);
 
@@ -469,12 +457,10 @@ lfc_read(RelFileNode rnode, ForkNumber forkNum, BlockNumber blkno,
 
 	if (lfc_size_limit == 0) /* fast exit if file cache is disabled */
 		return false;
-#if PG_VERSION_NUM >= 160000
-	InitBufferTag(&tag, &rnode, forkNum, (blkno & ~(BLOCKS_PER_CHUNK-1)));
-#else
-	INIT_BUFFERTAG(tag, rnode, forkNum, (blkno & ~(BLOCKS_PER_CHUNK-1)));
-#endif
 
+	tag.rnode = rnode;
+	tag.forkNum = forkNum;
+	tag.blockNum = blkno & ~(BLOCKS_PER_CHUNK-1);
 	hash = get_hash_value(lfc_hash, &tag);
 
 	LWLockAcquire(lfc_lock, LW_EXCLUSIVE);
@@ -540,12 +526,9 @@ lfc_write(RelFileNode rnode, ForkNumber forkNum, BlockNumber blkno,
 	if (lfc_size_limit == 0) /* fast exit if file cache is disabled */
 		return;
 
-#if PG_VERSION_NUM >= 160000
-	InitBufferTag(&tag, &rnode, forkNum, (blkno & ~(BLOCKS_PER_CHUNK-1)));
-#else
-	INIT_BUFFERTAG(tag, rnode, forkNum, (blkno & ~(BLOCKS_PER_CHUNK-1)));
-#endif
-
+	tag.rnode = rnode;
+	tag.forkNum = forkNum;
+	tag.blockNum = blkno & ~(BLOCKS_PER_CHUNK-1);
 	hash = get_hash_value(lfc_hash, &tag);
 
 	LWLockAcquire(lfc_lock, LW_EXCLUSIVE);
@@ -739,16 +722,9 @@ local_cache_pages(PG_FUNCTION_ARGS)
 				if (entry->bitmap[i >> 5] & (1 << (i & 31)))
 				{
 					fctx->record[n_pages].pageoffs = entry->offset*BLOCKS_PER_CHUNK + i;
-
-#if PG_VERSION_NUM >= 160000
-					fctx->record[n_pages].relfilenode =  entry->key.relNumber;
-					fctx->record[n_pages].reltablespace = entry->key.spcOid;
-					fctx->record[n_pages].reldatabase = entry->key.dbOid;
-#else
 					fctx->record[n_pages].relfilenode = entry->key.rnode.relNode;
 					fctx->record[n_pages].reltablespace = entry->key.rnode.spcNode;
 					fctx->record[n_pages].reldatabase = entry->key.rnode.dbNode;
-#endif
 					fctx->record[n_pages].forknum = entry->key.forkNum;
 					fctx->record[n_pages].blocknum = entry->key.blockNum + i;
 					fctx->record[n_pages].accesscount = entry->access_count;

pgxn/neon/libpagestore.c

@@ -34,6 +34,7 @@
 
 #define PageStoreTrace DEBUG5
 
+#define MAX_RECONNECT_ATTEMPTS 5
 #define RECONNECT_INTERVAL_USEC 1000000
 
 bool		connected = false;
@@ -54,15 +55,13 @@ int32		max_cluster_size;
 char	   *page_server_connstring;
 char	   *neon_auth_token;
 
-int			readahead_buffer_size = 128;
+int			n_unflushed_requests = 0;
 int			flush_every_n_requests = 8;
-
-int			n_reconnect_attempts = 0;
-int			max_reconnect_attempts = 60;
+int			readahead_buffer_size = 128;
 
 bool	(*old_redo_read_buffer_filter) (XLogReaderState *record, uint8 block_id) = NULL;
 
-static bool pageserver_flush(void);
+static void pageserver_flush(void);
 
 static bool
 pageserver_connect(int elevel)
@@ -233,17 +232,16 @@ pageserver_disconnect(void)
 	}
 }
 
-static bool
+static void
 pageserver_send(NeonRequest * request)
 {
 	StringInfoData req_buff;
+	int n_reconnect_attempts = 0;
 
 	/* If the connection was lost for some reason, reconnect */
 	if (connected && PQstatus(pageserver_conn) == CONNECTION_BAD)
-	{
-		neon_log(LOG, "pageserver_send disconnect bad connection");
 		pageserver_disconnect();
-	}
+
 
 	req_buff = nm_pack_request(request);
 
@@ -254,36 +252,53 @@ pageserver_send(NeonRequest * request)
 	 * See https://github.com/neondatabase/neon/issues/1138
 	 * So try to reestablish connection in case of failure.
 	 */
-	if (!connected)
+	while (true)
 	{
-		while (!pageserver_connect(n_reconnect_attempts < max_reconnect_attempts ? LOG : ERROR))
+		if (!connected)
 		{
-			n_reconnect_attempts += 1;
-			pg_usleep(RECONNECT_INTERVAL_USEC);
+			if (!pageserver_connect(n_reconnect_attempts < MAX_RECONNECT_ATTEMPTS ? LOG : ERROR))
+			{
+				n_reconnect_attempts += 1;
+				pg_usleep(RECONNECT_INTERVAL_USEC);
+				continue;
+			}
 		}
-		n_reconnect_attempts = 0;
-	}
 
-	/*
-	 * Send request.
-	 *
-	 * In principle, this could block if the output buffer is full, and we
-	 * should use async mode and check for interrupts while waiting. In
-	 * practice, our requests are small enough to always fit in the output and
-	 * TCP buffer.
-	 */
-	if (PQputCopyData(pageserver_conn, req_buff.data, req_buff.len) <= 0)
-	{
-		char	   *msg = pchomp(PQerrorMessage(pageserver_conn));
-		pageserver_disconnect();
-		neon_log(LOG, "pageserver_send disconnect because failed to send page request (try to reconnect): %s", msg);
-		pfree(msg);
-		pfree(req_buff.data);
-		return false;
+		/*
+		 * Send request.
+		 *
+		 * In principle, this could block if the output buffer is full, and we
+		 * should use async mode and check for interrupts while waiting. In
+		 * practice, our requests are small enough to always fit in the output and
+		 * TCP buffer.
+		 */
+		if (PQputCopyData(pageserver_conn, req_buff.data, req_buff.len) <= 0)
+		{
+			char	   *msg = pchomp(PQerrorMessage(pageserver_conn));
+			if (n_reconnect_attempts < MAX_RECONNECT_ATTEMPTS)
+			{
+				neon_log(LOG, "failed to send page request (try to reconnect): %s", msg);
+				if (n_reconnect_attempts != 0) /* do not sleep before first reconnect attempt, assuming that pageserver is already restarted */
+					pg_usleep(RECONNECT_INTERVAL_USEC);
+				n_reconnect_attempts += 1;
+				continue;
+			}
+			else
+			{
+				pageserver_disconnect();
+				neon_log(ERROR, "failed to send page request: %s", msg);
+			}
+		}
+		break;
 	}
 
 	pfree(req_buff.data);
 
+	n_unflushed_requests++;
+
+	if (flush_every_n_requests > 0 && n_unflushed_requests >= flush_every_n_requests)
+		pageserver_flush();
+
 	if (message_level_is_interesting(PageStoreTrace))
 	{
 		char	   *msg = nm_to_string((NeonMessage *) request);
@@ -291,7 +306,6 @@ pageserver_send(NeonRequest * request)
 		neon_log(PageStoreTrace, "sent request: %s", msg);
 		pfree(msg);
 	}
-	return true;
 }
 
 static NeonResponse *
@@ -326,25 +340,16 @@ pageserver_receive(void)
 		}
 		else if (rc == -1)
 		{
-			neon_log(LOG, "pageserver_receive disconnect because call_PQgetCopyData returns -1: %s", pchomp(PQerrorMessage(pageserver_conn)));
 			pageserver_disconnect();
 			resp = NULL;
 		}
 		else if (rc == -2)
-		{
-			char* msg = pchomp(PQerrorMessage(pageserver_conn));
-			pageserver_disconnect();
-			neon_log(ERROR, "pageserver_receive disconnect because could not read COPY data: %s", msg);
-		}
+			neon_log(ERROR, "could not read COPY data: %s", pchomp(PQerrorMessage(pageserver_conn)));
 		else
-		{
-			pageserver_disconnect();
-			neon_log(ERROR, "pageserver_receive disconnect because unexpected PQgetCopyData return value: %d", rc);
-		}
+			neon_log(ERROR, "unexpected PQgetCopyData return value: %d", rc);
 	}
 	PG_CATCH();
 	{
-		neon_log(LOG, "pageserver_receive disconnect due to caught exception");
 		pageserver_disconnect();
 		PG_RE_THROW();
 	}
@@ -354,25 +359,21 @@ pageserver_receive(void)
 }
 
 
-static bool
+static void
 pageserver_flush(void)
 {
 	if (!connected)
 	{
 		neon_log(WARNING, "Tried to flush while disconnected");
 	}
-	else
+	else if (PQflush(pageserver_conn))
 	{
-		if (PQflush(pageserver_conn))
-		{
-			char	   *msg = pchomp(PQerrorMessage(pageserver_conn));
-			pageserver_disconnect();
-			neon_log(LOG, "pageserver_flush disconnect because failed to flush page requests: %s", msg);
-			pfree(msg);
-			return false;
-		}
+		char	   *msg = pchomp(PQerrorMessage(pageserver_conn));
+
+		pageserver_disconnect();
+		neon_log(ERROR, "failed to flush page requests: %s", msg);
 	}
-	return true;
+	n_unflushed_requests = 0;
 }
 
 page_server_api api = {
@@ -438,14 +439,6 @@ pg_init_libpagestore(void)
 							PGC_USERSET,
 							0,	/* no flags required */
 							NULL, NULL, NULL);
-	DefineCustomIntVariable("neon.max_reconnect_attempts",
-							"Maximal attempts to reconnect to pages server (with 1 second timeout)",
-							NULL,
-							&max_reconnect_attempts,
-							10, 0, INT_MAX,
-							PGC_USERSET,
-							0,
-							NULL, NULL, NULL);
 	DefineCustomIntVariable("neon.readahead_buffer_size",
 							"number of prefetches to buffer",
 							"This buffer is used to hold and manage prefetched "

pgxn/neon/neon--1.0.sql

@@ -32,3 +32,7 @@ CREATE VIEW local_cache AS
 	SELECT P.* FROM local_cache_pages() AS P
 	(pageoffs int8, relfilenode oid, reltablespace oid, reldatabase oid,
 	 relforknumber int2, relblocknumber int8, accesscount int4);
+
+CREATE FUNCTION copy_from(conninfo cstring) RETURNS BIGINT
+AS 'MODULE_PATHNAME', 'copy_from'
+LANGUAGE C;

pgxn/neon/neon.c

@@ -13,20 +13,32 @@
 
 #include "access/xact.h"
 #include "access/xlog.h"
+#include "access/relation.h"
+#include "access/xloginsert.h"
 #include "storage/buf_internals.h"
 #include "storage/bufmgr.h"
 #include "catalog/pg_type.h"
+#include "catalog/namespace.h"
 #include "replication/walsender.h"
 #include "funcapi.h"
+#include "miscadmin.h"
 #include "access/htup_details.h"
 #include "utils/pg_lsn.h"
 #include "utils/guc.h"
+#include "utils/wait_event.h"
+#include "utils/rel.h"
+#include "utils/varlena.h"
+#include "utils/builtins.h"
 
 #include "neon.h"
 #include "walproposer.h"
 #include "pagestore_client.h"
 #include "control_plane_connector.h"
 
+#include "libpq-fe.h"
+#include "libpq/pqformat.h"
+#include "libpq/libpq.h"
+
 PG_MODULE_MAGIC;
 void		_PG_init(void);
 
@@ -46,6 +58,7 @@ _PG_init(void)
 PG_FUNCTION_INFO_V1(pg_cluster_size);
 PG_FUNCTION_INFO_V1(backpressure_lsns);
 PG_FUNCTION_INFO_V1(backpressure_throttling_time);
+PG_FUNCTION_INFO_V1(copy_from);
 
 Datum
 pg_cluster_size(PG_FUNCTION_ARGS)
@@ -91,3 +104,281 @@ backpressure_throttling_time(PG_FUNCTION_ARGS)
 {
 	PG_RETURN_UINT64(BackpressureThrottlingTime());
 }
+
+
+#define N_RAW_PAGE_COLUMNS 4
+#define COPY_FETCH_COUNT   16
+
+
+static void
+report_error(int elevel, PGresult *res, PGconn *conn,
+			 bool clear, const char *sql)
+{
+	/* If requested, PGresult must be released before leaving this function. */
+	PG_TRY();
+	{
+		char	   *diag_sqlstate = PQresultErrorField(res, PG_DIAG_SQLSTATE);
+		char	   *message_primary = PQresultErrorField(res, PG_DIAG_MESSAGE_PRIMARY);
+		char	   *message_detail = PQresultErrorField(res, PG_DIAG_MESSAGE_DETAIL);
+		char	   *message_hint = PQresultErrorField(res, PG_DIAG_MESSAGE_HINT);
+		char	   *message_context = PQresultErrorField(res, PG_DIAG_CONTEXT);
+		int			sqlstate;
+
+		if (diag_sqlstate)
+			sqlstate = MAKE_SQLSTATE(diag_sqlstate[0],
+									 diag_sqlstate[1],
+									 diag_sqlstate[2],
+									 diag_sqlstate[3],
+									 diag_sqlstate[4]);
+		else
+			sqlstate = ERRCODE_CONNECTION_FAILURE;
+
+		/*
+		 * If we don't get a message from the PGresult, try the PGconn.  This
+		 * is needed because for connection-level failures, PQexec may just
+		 * return NULL, not a PGresult at all.
+		 */
+		if (message_primary == NULL)
+			message_primary = pchomp(PQerrorMessage(conn));
+
+		ereport(elevel,
+				(errcode(sqlstate),
+				 (message_primary != NULL && message_primary[0] != '\0') ?
+				 errmsg_internal("%s", message_primary) :
+				 errmsg("could not obtain message string for remote error"),
+				 message_detail ? errdetail_internal("%s", message_detail) : 0,
+				 message_hint ? errhint("%s", message_hint) : 0,
+				 message_context ? errcontext("%s", message_context) : 0,
+				 sql ? errcontext("remote SQL command: %s", sql) : 0));
+	}
+	PG_FINALLY();
+	{
+		if (clear)
+			PQclear(res);
+	}
+	PG_END_TRY();
+}
+
+static PGresult *
+get_result(PGconn *conn, const char *query)
+{
+	PGresult   *volatile last_res = NULL;
+
+	/* In what follows, do not leak any PGresults on an error. */
+	PG_TRY();
+	{
+		for (;;)
+		{
+			PGresult   *res;
+
+			while (PQisBusy(conn))
+			{
+				int			wc;
+
+				/* Sleep until there's something to do */
+				wc = WaitLatchOrSocket(MyLatch,
+									   WL_LATCH_SET | WL_SOCKET_READABLE |
+									   WL_EXIT_ON_PM_DEATH,
+									   PQsocket(conn),
+									   -1L, PG_WAIT_EXTENSION);
+				ResetLatch(MyLatch);
+
+				CHECK_FOR_INTERRUPTS();
+
+				/* Data available in socket? */
+				if (wc & WL_SOCKET_READABLE)
+				{
+					if (!PQconsumeInput(conn))
+						report_error(ERROR, NULL, conn, false, query);
+				}
+			}
+
+			res = PQgetResult(conn);
+			if (res == NULL)
+				break;			/* query is complete */
+
+			PQclear(last_res);
+			last_res = res;
+		}
+	}
+	PG_CATCH();
+	{
+		PQclear(last_res);
+		PG_RE_THROW();
+	}
+	PG_END_TRY();
+
+	return last_res;
+}
+
+#define CREATE_COPYDATA_FUNC "\
+create or replace function copydata() returns setof record as $$ \
+declare \
+    relsize integer; \
+    total_relsize integer; \
+	content bytea; \
+	r record; \
+	fork text; \
+	relname text; \
+	pagesize integer; \
+begin \
+	pagesize = current_setting('block_size'); \
+	for r in select oid,reltoastrelid from pg_class where relnamespace not in (select oid from pg_namespace where nspname in ('pg_catalog','pg_toast','information_schema')) \
+	loop \
+		relname = r.oid::regclass::text; \
+        total_relsize = 0; \
+	    foreach fork in array array['main','vm','fsm'] \
+		loop \
+		    relsize = pg_relation_size(r.oid, fork)/pagesize; \
+			total_relsize = total_relsize + relsize; \
+	        for p in 1..relsize \
+		    loop \
+			    content = get_raw_page(relname, fork, p-1); \
+				return next row(relname,fork,p-1,content); \
+			end loop; \
+		end loop; \
+        if total_relsize <> 0 and r.reltoastrelid <> 0 then \
+            foreach relname in array array ['pg_toast.pg_toast_'||r.oid, 'pg_toast.pg_toast_'||r.oid||'_index'] \
+			loop \
+		    	foreach fork in array array['main','vm','fsm'] \
+				loop \
+			    	relsize = pg_relation_size(relname, fork)/pagesize; \
+	        		for p in 1..relsize \
+		    		loop \
+			    		content = get_raw_page(relname, fork, p-1); \
+						return next row(relname,fork,p-1,content); \
+					end loop; \
+				end loop; \
+			end loop; \
+        end if; \
+	end loop; \
+end; \
+$$ language plpgsql"
+
+Datum
+copy_from(PG_FUNCTION_ARGS)
+{
+	char const* conninfo = PG_GETARG_CSTRING(0);
+	PGconn* conn;
+	char const* declare_cursor = "declare copy_data_cursor no scroll cursor for select * from copydata() as raw_page(relid text, fork text, blkno integer, content bytea)";
+	char* fetch_cursor = psprintf("fetch forward %d copy_data_cursor", COPY_FETCH_COUNT);
+	char const* close_cursor = "close copy_data_cursor";
+	char const* vacuum_freeze = "vacuum freeze";
+	char   *content;
+	char const* relname;
+	BlockNumber blkno;
+	ForkNumber forknum;
+	BlockNumber prev_blkno = InvalidBlockNumber;
+	RangeVar   *relrv;
+	Relation rel = NULL;
+	BlockNumber rel_size;
+	int64_t total = 0;
+	PGresult   *res;
+	char blkno_buf[4];
+	int n_tuples;
+	Buffer buf;
+	char* toast_rel_name;
+	Oid relid = InvalidOid;
+
+	/* Connect to the source database */
+	conn = PQconnectdb(conninfo);
+	if (!conn || PQstatus(conn) != CONNECTION_OK)
+		ereport(ERROR,
+				(errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
+				 errmsg("could not connect to server \"%s\"",
+						conninfo),
+				 errdetail_internal("%s", pchomp(PQerrorMessage(conn)))));
+
+	/* First create store procedure (assumes that pageinspector extension is already installed) */
+	res = PQexec(conn, CREATE_COPYDATA_FUNC);
+	if (res == NULL || PQresultStatus(res) != PGRES_COMMAND_OK)
+		report_error(ERROR, res, conn, true, CREATE_COPYDATA_FUNC);
+	PQclear(res);
+
+	/* Freeze all tables to prevent problems with XID mapping */
+	res = PQexec(conn, vacuum_freeze);
+	if (res == NULL || PQresultStatus(res) != PGRES_COMMAND_OK)
+		report_error(ERROR, res, conn, true, vacuum_freeze);
+	PQclear(res);
+
+	/* Start transaction to use cursor */
+	res = PQexec(conn, "BEGIN");
+	if (res == NULL || PQresultStatus(res) != PGRES_COMMAND_OK)
+		report_error(ERROR, res, conn, true, "BEGIN");
+	PQclear(res);
+
+	/* Declare cursor (we have to use cursor to avoid materializing all database in memory) */
+	res = PQexec(conn, declare_cursor);
+	if (res == NULL || PQresultStatus(res) != PGRES_COMMAND_OK)
+		report_error(ERROR, res, conn, true, declare_cursor);
+	PQclear(res);
+
+	/* Get database data */
+	while ((res = PQexecParams(conn, fetch_cursor, 0, NULL, NULL, NULL, NULL, 1)) != NULL)
+	{
+		if (PQresultStatus(res) != PGRES_TUPLES_OK)
+			report_error(ERROR, res, conn, true, fetch_cursor);
+
+		n_tuples = PQntuples(res);
+		if (PQnfields(res) != 4)
+			elog(ERROR, "unexpected result from copydata()");
+
+		for (int i = 0; i < n_tuples; i++)
+		{
+			relname = PQgetvalue(res, i, 0);
+			forknum = forkname_to_number(PQgetvalue(res, i, 1));
+			memcpy(&blkno, PQgetvalue(res, i, 2), sizeof(BlockNumber));
+			blkno = pg_ntoh32(blkno);
+			content = (char*)PQgetvalue(res, i, 3);
+
+			if (blkno <= prev_blkno)
+			{
+				if (forknum == MAIN_FORKNUM)
+				{
+					char* dst_rel_name = strncmp(relname, "pg_toast.", 9) == 0
+						/* Construct correct TOAST table name */
+						? psprintf("pg_toast.pg_toast_%u%s",
+								   relid,
+								   strcmp(relname + strlen(relname) - 5, "index") == 0 ? "_index" : "")
+						: (char*)relname;
+					if (rel)
+						relation_close(rel, AccessExclusiveLock);
+					relrv = makeRangeVarFromNameList(textToQualifiedNameList(cstring_to_text(dst_rel_name)));
+					rel = relation_openrv(relrv, AccessExclusiveLock);
+					if (dst_rel_name != relname)
+						pfree(dst_rel_name);
+					else
+						relid = RelationGetRelid(rel);
+				}
+				rel_size = RelationGetNumberOfBlocksInFork(rel, forknum);
+			}
+			buf = ReadBufferExtended(rel, forknum, blkno < rel_size ? blkno : P_NEW, RBM_ZERO_AND_LOCK, NULL);
+			MarkBufferDirty(buf);
+			memcpy(BufferGetPage(buf), content, BLCKSZ);
+			log_newpage_buffer(buf, forknum == MAIN_FORKNUM);
+			UnlockReleaseBuffer(buf);
+
+			total += 1;
+			prev_blkno = blkno;
+		}
+		PQclear(res);
+		if (n_tuples < COPY_FETCH_COUNT)
+			break;
+	}
+	res = PQexec(conn, close_cursor);
+	if (res == NULL || PQresultStatus(res) != PGRES_COMMAND_OK)
+		report_error(ERROR, res, conn, true, close_cursor);
+	PQclear(res);
+
+	if (rel)
+		relation_close(rel, AccessExclusiveLock);
+
+	/* Complete transaction */
+	res = PQexec(conn, "END");
+	if (res == NULL || PQresultStatus(res) != PGRES_COMMAND_OK)
+		report_error(ERROR, res, conn, true, "END");
+	PQclear(res);
+
+	PQfinish(conn);
+	PG_RETURN_INT64(total);
+}

pgxn/neon/pagestore_client.h

@@ -16,11 +16,7 @@
 #include "postgres.h"
 
 #include "access/xlogdefs.h"
-#if PG_VERSION_NUM >= 160000
-#include "storage/relfilelocator.h"
-#else
 #include "storage/relfilenode.h"
-#endif
 #include "storage/block.h"
 #include "storage/smgr.h"
 #include "lib/stringinfo.h"
@@ -29,34 +25,6 @@
 
 #include "pg_config.h"
 
-// This is a hack to avoid too many ifdefs in the function definitions.
-#if PG_VERSION_NUM >= 160000
-typedef RelFileLocator RelFileNode;
-typedef RelFileLocatorBackend RelFileNodeBackend;
-#define RelFileNodeBackendIsTemp RelFileLocatorBackendIsTemp
-#endif
-
-#if PG_VERSION_NUM >= 160000
-#define RelnGetRnode(reln) (reln->smgr_rlocator.locator)
-#define RnodeGetSpcOid(rnode) (rnode.spcOid)
-#define RnodeGetDbOid(rnode) (rnode.dbOid)
-#define RnodeGetRelNumber(rnode) (rnode.relNumber)
-
-#define BufTagGetRnode(tag) (BufTagGetRelFileLocator(&tag))
-#else
-#define RelnGetRnode(reln) (reln->smgr_rnode.node)
-#define RnodeGetSpcOid(rnode) (rnode.spcNode)
-#define RnodeGetDbOid(rnode) (rnode.dbNode)
-#define RnodeGetRelNumber(rnode) (rnode.relNode)
-
-#define BufTagGetRnode(tag) (tag.rnode)
-
-#endif
-
-#define RelnGetSpcOid(reln) (RnodeGetRelNumber(RelnGetRnode(reln)))
-#define RelnGetDbOid(reln) (RnodeGetDbOid(RelnGetRnode(reln)))
-#define RelnGetRelNumber(reln) (RnodeGetRelNumber(RelnGetRnode(reln)))
-
 typedef enum
 {
 	/* pagestore_client -> pagestore */
@@ -117,7 +85,7 @@ typedef struct
 typedef struct
 {
 	NeonRequest req;
-	Oid			dbOid;
+	Oid			dbNode;
 }			NeonDbSizeRequest;
 
 typedef struct
@@ -177,9 +145,9 @@ extern char *nm_to_string(NeonMessage * msg);
 
 typedef struct
 {
-	bool		(*send) (NeonRequest * request);
+	void		(*send) (NeonRequest * request);
 	NeonResponse *(*receive) (void);
-	bool		(*flush) (void);
+	void		(*flush) (void);
 }			page_server_api;
 
 extern void prefetch_on_ps_disconnect(void);

pgxn/neon/pagestore_smgr.c

@@ -58,11 +58,7 @@
 #include "postmaster/autovacuum.h"
 #include "replication/walsender.h"
 #include "storage/bufmgr.h"
-#if PG_VERSION_NUM >= 160000
-#include "storage/relfilelocator.h"
-#else
 #include "storage/relfilenode.h"
-#endif
 #include "storage/buf_internals.h"
 #include "storage/smgr.h"
 #include "storage/md.h"
@@ -74,8 +70,6 @@
 #include "access/xlogrecovery.h"
 #endif
 
-
-
 /*
  * If DEBUG_COMPARE_LOCAL is defined, we pass through all the SMGR API
  * calls to md.c, and *also* do the calls to the Page Server. On every
@@ -92,10 +86,7 @@
 static char *hexdump_page(char *page);
 #endif
 
-
-#define IS_LOCAL_REL(reln) (RelnGetDbOid(reln) != 0 && RelnGetRelNumber(reln) > FirstNormalObjectId)
-
-
+#define IS_LOCAL_REL(reln) (reln->smgr_rnode.node.dbNode != 0 && reln->smgr_rnode.node.relNode > FirstNormalObjectId)
 
 const int	SmgrTrace = DEBUG5;
 
@@ -193,13 +184,7 @@ typedef struct PrfHashEntry {
 	sizeof(BufferTag) \
 )
 
-
-#if PG_VERSION_NUM >= 160000
-#define SH_EQUAL(tb, a, b)	(BufferTagsEqual(&((a)->buftag),&((b)->buftag)))
-#else
 #define SH_EQUAL(tb, a, b)	(BUFFERTAGS_EQUAL((a)->buftag, (b)->buftag))
-#endif
-
 #define SH_SCOPE			static inline
 #define SH_DEFINE
 #define SH_DECLARE
@@ -504,8 +489,7 @@ prefetch_wait_for(uint64 ring_index)
 	if (MyPState->ring_flush <= ring_index &&
 		MyPState->ring_unused > MyPState->ring_flush)
 	{
-		if (!page_server->flush())
-			return false;
+		page_server->flush();
 		MyPState->ring_flush = MyPState->ring_unused;
 	}
 
@@ -649,7 +633,7 @@ prefetch_do_request(PrefetchRequest *slot, bool *force_latest, XLogRecPtr *force
 		.req.tag = T_NeonGetPageRequest,
 		.req.latest = false,
 		.req.lsn = 0,
-		.rnode = BufTagGetRnode(slot->buftag),
+		.rnode = slot->buftag.rnode,
 		.forknum = slot->buftag.forkNum,
 		.blkno = slot->buftag.blockNum,
 	};
@@ -664,7 +648,7 @@ prefetch_do_request(PrefetchRequest *slot, bool *force_latest, XLogRecPtr *force
 	{
 		XLogRecPtr lsn = neon_get_request_lsn(
 			&request.req.latest,
-			BufTagGetRnode(slot->buftag),
+			slot->buftag.rnode,
 			slot->buftag.forkNum,
 			slot->buftag.blockNum
 		);
@@ -682,7 +666,7 @@ prefetch_do_request(PrefetchRequest *slot, bool *force_latest, XLogRecPtr *force
 		 * smaller than the current WAL insert/redo pointer, which is already
 		 * larger than this prefetch_lsn. So in any case, that would
 		 * invalidate this cache.
-		 *
+		 * 
 		 * The best LSN to use for effective_request_lsn would be
 		 * XLogCtl->Insert.RedoRecPtr, but that's expensive to access.
 		 */
@@ -693,8 +677,7 @@ prefetch_do_request(PrefetchRequest *slot, bool *force_latest, XLogRecPtr *force
 
 	Assert(slot->response == NULL);
 	Assert(slot->my_ring_index == MyPState->ring_unused);
-
-	while (!page_server->send((NeonRequest *) &request));
+	page_server->send((NeonRequest *) &request);
 
 	/* update prefetch state */
 	MyPState->n_requests_inflight += 1;
@@ -704,7 +687,6 @@ prefetch_do_request(PrefetchRequest *slot, bool *force_latest, XLogRecPtr *force
 	/* update slot state */
 	slot->status = PRFS_REQUESTED;
 
-
 	prfh_insert(MyPState->prf_hash, slot, &found);
 	Assert(!found);
 }
@@ -744,11 +726,8 @@ prefetch_register_buffer(BufferTag tag, bool *force_latest, XLogRecPtr *force_ls
 		Assert(slot->status != PRFS_UNUSED);
 		Assert(MyPState->ring_last <= ring_index &&
 			   ring_index < MyPState->ring_unused);
-#if PG_VERSION_NUM >= 160000
-		Assert(BufferTagsEqual(&slot->buftag, &tag));
-#else
 		Assert(BUFFERTAGS_EQUAL(slot->buftag, tag));
-#endif
+
 		/*
 		 * If we want a specific lsn, we do not accept requests that were made
 		 * with a potentially different LSN.
@@ -764,7 +743,6 @@ prefetch_register_buffer(BufferTag tag, bool *force_latest, XLogRecPtr *force_ls
 					prefetch_set_unused(ring_index);
 					entry = NULL;
 				}
-
 			}
 			/* if we don't want the latest version, only accept requests with the exact same LSN */
 			else
@@ -778,23 +756,20 @@ prefetch_register_buffer(BufferTag tag, bool *force_latest, XLogRecPtr *force_ls
 			}
 		}
 
-		if (entry != NULL)
+		/*
+		 * We received a prefetch for a page that was recently read and
+		 * removed from the buffers. Remove that request from the buffers.
+		 */
+		else if (slot->status == PRFS_TAG_REMAINS)
 		{
-			/*
-			 * We received a prefetch for a page that was recently read and
-			 * removed from the buffers. Remove that request from the buffers.
-			 */
-			if (slot->status == PRFS_TAG_REMAINS)
-			{
-				prefetch_set_unused(ring_index);
-				entry = NULL;
-			}
-			else
-			{
-				/* The buffered request is good enough, return that index */
-				pgBufferUsage.prefetch.duplicates++;
-				return ring_index;
-			}
+			prefetch_set_unused(ring_index);
+			entry = NULL;
+		}
+		else
+		{
+			/* The buffered request is good enough, return that index */
+			pgBufferUsage.prefetch.duplicates++;
+			return ring_index;
 		}
 	}
 
@@ -884,7 +859,8 @@ page_server_request(void const *req)
 {
 	NeonResponse* resp;
 	do {
-		while (!page_server->send((NeonRequest *) req) || !page_server->flush());
+		page_server->send((NeonRequest *) req);
+		page_server->flush();
 		MyPState->ring_flush = MyPState->ring_unused;
 		consume_prefetch_responses();
 		resp = page_server->receive();
@@ -911,9 +887,9 @@ nm_pack_request(NeonRequest * msg)
 
 				pq_sendbyte(&s, msg_req->req.latest);
 				pq_sendint64(&s, msg_req->req.lsn);
-				pq_sendint32(&s, RnodeGetSpcOid(msg_req->rnode));
-				pq_sendint32(&s, RnodeGetDbOid(msg_req->rnode));
-				pq_sendint32(&s, RnodeGetRelNumber(msg_req->rnode));
+				pq_sendint32(&s, msg_req->rnode.spcNode);
+				pq_sendint32(&s, msg_req->rnode.dbNode);
+				pq_sendint32(&s, msg_req->rnode.relNode);
 				pq_sendbyte(&s, msg_req->forknum);
 
 				break;
@@ -924,9 +900,9 @@ nm_pack_request(NeonRequest * msg)
 
 				pq_sendbyte(&s, msg_req->req.latest);
 				pq_sendint64(&s, msg_req->req.lsn);
-				pq_sendint32(&s, RnodeGetSpcOid(msg_req->rnode));
-				pq_sendint32(&s, RnodeGetDbOid(msg_req->rnode));
-				pq_sendint32(&s, RnodeGetRelNumber(msg_req->rnode));
+				pq_sendint32(&s, msg_req->rnode.spcNode);
+				pq_sendint32(&s, msg_req->rnode.dbNode);
+				pq_sendint32(&s, msg_req->rnode.relNode);
 				pq_sendbyte(&s, msg_req->forknum);
 
 				break;
@@ -937,7 +913,7 @@ nm_pack_request(NeonRequest * msg)
 
 				pq_sendbyte(&s, msg_req->req.latest);
 				pq_sendint64(&s, msg_req->req.lsn);
-				pq_sendint32(&s, msg_req->dbOid);
+				pq_sendint32(&s, msg_req->dbNode);
 
 				break;
 			}
@@ -947,9 +923,9 @@ nm_pack_request(NeonRequest * msg)
 
 				pq_sendbyte(&s, msg_req->req.latest);
 				pq_sendint64(&s, msg_req->req.lsn);
-				pq_sendint32(&s, RnodeGetSpcOid(msg_req->rnode));
-				pq_sendint32(&s, RnodeGetDbOid(msg_req->rnode));
-				pq_sendint32(&s, RnodeGetRelNumber(msg_req->rnode));
+				pq_sendint32(&s, msg_req->rnode.spcNode);
+				pq_sendint32(&s, msg_req->rnode.dbNode);
+				pq_sendint32(&s, msg_req->rnode.relNode);
 				pq_sendbyte(&s, msg_req->forknum);
 				pq_sendint32(&s, msg_req->blkno);
 
@@ -1082,9 +1058,9 @@ nm_to_string(NeonMessage * msg)
 
 				appendStringInfoString(&s, "{\"type\": \"NeonExistsRequest\"");
 				appendStringInfo(&s, ", \"rnode\": \"%u/%u/%u\"",
-								 RnodeGetSpcOid(msg_req->rnode),
-								 RnodeGetDbOid(msg_req->rnode),
-								 RnodeGetRelNumber(msg_req->rnode));
+								 msg_req->rnode.spcNode,
+								 msg_req->rnode.dbNode,
+								 msg_req->rnode.relNode);
 				appendStringInfo(&s, ", \"forknum\": %d", msg_req->forknum);
 				appendStringInfo(&s, ", \"lsn\": \"%X/%X\"", LSN_FORMAT_ARGS(msg_req->req.lsn));
 				appendStringInfo(&s, ", \"latest\": %d", msg_req->req.latest);
@@ -1098,9 +1074,9 @@ nm_to_string(NeonMessage * msg)
 
 				appendStringInfoString(&s, "{\"type\": \"NeonNblocksRequest\"");
 				appendStringInfo(&s, ", \"rnode\": \"%u/%u/%u\"",
-								 RnodeGetSpcOid(msg_req->rnode),
-								 RnodeGetDbOid(msg_req->rnode),
-								 RnodeGetRelNumber(msg_req->rnode));
+								 msg_req->rnode.spcNode,
+								 msg_req->rnode.dbNode,
+								 msg_req->rnode.relNode);
 				appendStringInfo(&s, ", \"forknum\": %d", msg_req->forknum);
 				appendStringInfo(&s, ", \"lsn\": \"%X/%X\"", LSN_FORMAT_ARGS(msg_req->req.lsn));
 				appendStringInfo(&s, ", \"latest\": %d", msg_req->req.latest);
@@ -1114,9 +1090,9 @@ nm_to_string(NeonMessage * msg)
 
 				appendStringInfoString(&s, "{\"type\": \"NeonGetPageRequest\"");
 				appendStringInfo(&s, ", \"rnode\": \"%u/%u/%u\"",
-								 RnodeGetSpcOid(msg_req->rnode),
-								 RnodeGetDbOid(msg_req->rnode),
-								 RnodeGetRelNumber(msg_req->rnode));
+								 msg_req->rnode.spcNode,
+								 msg_req->rnode.dbNode,
+								 msg_req->rnode.relNode);
 				appendStringInfo(&s, ", \"forknum\": %d", msg_req->forknum);
 				appendStringInfo(&s, ", \"blkno\": %u", msg_req->blkno);
 				appendStringInfo(&s, ", \"lsn\": \"%X/%X\"", LSN_FORMAT_ARGS(msg_req->req.lsn));
@@ -1129,7 +1105,7 @@ nm_to_string(NeonMessage * msg)
 				NeonDbSizeRequest *msg_req = (NeonDbSizeRequest *) msg;
 
 				appendStringInfoString(&s, "{\"type\": \"NeonDbSizeRequest\"");
-				appendStringInfo(&s, ", \"dbnode\": \"%u\"", msg_req->dbOid);
+				appendStringInfo(&s, ", \"dbnode\": \"%u\"", msg_req->dbNode);
 				appendStringInfo(&s, ", \"lsn\": \"%X/%X\"", LSN_FORMAT_ARGS(msg_req->req.lsn));
 				appendStringInfo(&s, ", \"latest\": %d", msg_req->req.latest);
 				appendStringInfoChar(&s, '}');
@@ -1231,7 +1207,6 @@ static void
 neon_wallog_page(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, char *buffer, bool force)
 {
 	XLogRecPtr	lsn = PageGetLSN(buffer);
-	RelFileNode rnode = RelnGetRnode(reln);
 
 	if (ShutdownRequestPending)
 		return;
@@ -1251,16 +1226,15 @@ neon_wallog_page(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, ch
 		/* FSM is never WAL-logged and we don't care. */
 		XLogRecPtr	recptr;
 
-
-		recptr = log_newpage_copy(&rnode, forknum, blocknum, buffer, false);
+		recptr = log_newpage_copy(&reln->smgr_rnode.node, forknum, blocknum, buffer, false);
 		XLogFlush(recptr);
 		lsn = recptr;
 		ereport(SmgrTrace,
 				(errmsg("Page %u of relation %u/%u/%u.%u was force logged. Evicted at lsn=%X/%X",
 						blocknum,
-						RelnGetSpcOid(reln),
-						RelnGetDbOid(reln),
-						RelnGetRelNumber(reln),
+						reln->smgr_rnode.node.spcNode,
+						reln->smgr_rnode.node.dbNode,
+						reln->smgr_rnode.node.relNode,
 						forknum, LSN_FORMAT_ARGS(lsn))));
 	}
 	else if (lsn == InvalidXLogRecPtr)
@@ -1288,9 +1262,9 @@ neon_wallog_page(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, ch
 			ereport(SmgrTrace,
 					(errmsg("Page %u of relation %u/%u/%u.%u is all-zeros",
 							blocknum,
-							RelnGetSpcOid(reln),
-							RelnGetDbOid(reln),
-							RelnGetRelNumber(reln),
+							reln->smgr_rnode.node.spcNode,
+							reln->smgr_rnode.node.dbNode,
+							reln->smgr_rnode.node.relNode,
 							forknum)));
 		}
 		else if (PageIsEmptyHeapPage(buffer))
@@ -1298,9 +1272,9 @@ neon_wallog_page(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, ch
 			ereport(SmgrTrace,
 					(errmsg("Page %u of relation %u/%u/%u.%u is an empty heap page with no LSN",
 							blocknum,
-							RelnGetSpcOid(reln),
-							RelnGetDbOid(reln),
-							RelnGetRelNumber(reln),
+							reln->smgr_rnode.node.spcNode,
+							reln->smgr_rnode.node.dbNode,
+							reln->smgr_rnode.node.relNode,
 							forknum)));
 		}
 		else
@@ -1308,9 +1282,9 @@ neon_wallog_page(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, ch
 			ereport(PANIC,
 					(errmsg("Page %u of relation %u/%u/%u.%u is evicted with zero LSN",
 							blocknum,
-							RelnGetSpcOid(reln),
-							RelnGetDbOid(reln),
-							RelnGetRelNumber(reln),
+							reln->smgr_rnode.node.spcNode,
+							reln->smgr_rnode.node.dbNode,
+							reln->smgr_rnode.node.relNode,
 							forknum)));
 		}
 	}
@@ -1319,9 +1293,9 @@ neon_wallog_page(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, ch
 		ereport(SmgrTrace,
 				(errmsg("Page %u of relation %u/%u/%u.%u is already wal logged at lsn=%X/%X",
 						blocknum,
-						RelnGetSpcOid(reln),
-						RelnGetDbOid(reln),
-						RelnGetRelNumber(reln),
+						reln->smgr_rnode.node.spcNode,
+						reln->smgr_rnode.node.dbNode,
+						reln->smgr_rnode.node.relNode,
 						forknum, LSN_FORMAT_ARGS(lsn))));
 	}
 
@@ -1329,7 +1303,7 @@ neon_wallog_page(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, ch
 	 * Remember the LSN on this page. When we read the page again, we must
 	 * read the same or newer version of it.
 	 */
-	SetLastWrittenLSNForBlock(lsn, rnode, forknum, blocknum);
+	SetLastWrittenLSNForBlock(lsn, reln->smgr_rnode.node, forknum, blocknum);
 }
 
 /*
@@ -1479,7 +1453,6 @@ neon_exists(SMgrRelation reln, ForkNumber forkNum)
 	BlockNumber n_blocks;
 	bool		latest;
 	XLogRecPtr	request_lsn;
-	RelFileNode rnode = RelnGetRnode(reln);
 
 	switch (reln->smgr_relpersistence)
 	{
@@ -1506,7 +1479,7 @@ neon_exists(SMgrRelation reln, ForkNumber forkNum)
 			elog(ERROR, "unknown relpersistence '%c'", reln->smgr_relpersistence);
 	}
 
-	if (get_cached_relsize(RelnGetRnode(reln), forkNum, &n_blocks))
+	if (get_cached_relsize(reln->smgr_rnode.node, forkNum, &n_blocks))
 	{
 		return true;
 	}
@@ -1521,20 +1494,20 @@ neon_exists(SMgrRelation reln, ForkNumber forkNum)
 	 *
 	 * For now, handle that special case here.
 	 */
-	if (RelnGetSpcOid(reln) == 0 &&
-		RelnGetDbOid(reln) == 0 &&
-		RelnGetRelNumber(reln) == 0)
+	if (reln->smgr_rnode.node.spcNode == 0 &&
+		reln->smgr_rnode.node.dbNode == 0 &&
+		reln->smgr_rnode.node.relNode == 0)
 	{
 		return false;
 	}
 
-	request_lsn = neon_get_request_lsn(&latest, rnode, forkNum, REL_METADATA_PSEUDO_BLOCKNO);
+	request_lsn = neon_get_request_lsn(&latest, reln->smgr_rnode.node, forkNum, REL_METADATA_PSEUDO_BLOCKNO);
 	{
 		NeonExistsRequest request = {
 			.req.tag = T_NeonExistsRequest,
 			.req.latest = latest,
 			.req.lsn = request_lsn,
-			.rnode = rnode,
+			.rnode = reln->smgr_rnode.node,
 		.forknum = forkNum};
 
 		resp = page_server_request(&request);
@@ -1550,9 +1523,9 @@ neon_exists(SMgrRelation reln, ForkNumber forkNum)
 			ereport(ERROR,
 					(errcode(ERRCODE_IO_ERROR),
 					 errmsg("could not read relation existence of rel %u/%u/%u.%u from page server at lsn %X/%08X",
-							RelnGetSpcOid(reln),
-							RelnGetDbOid(reln),
-							RelnGetRelNumber(reln),
+							reln->smgr_rnode.node.spcNode,
+							reln->smgr_rnode.node.dbNode,
+							reln->smgr_rnode.node.relNode,
 							forkNum,
 							(uint32) (request_lsn >> 32), (uint32) request_lsn),
 					 errdetail("page server returned error: %s",
@@ -1574,8 +1547,6 @@ neon_exists(SMgrRelation reln, ForkNumber forkNum)
 void
 neon_create(SMgrRelation reln, ForkNumber forkNum, bool isRedo)
 {
-	RelFileNode rnode = RelnGetRnode(reln);
-
 	switch (reln->smgr_relpersistence)
 	{
 		case 0:
@@ -1594,8 +1565,9 @@ neon_create(SMgrRelation reln, ForkNumber forkNum, bool isRedo)
 	}
 
 	elog(SmgrTrace, "Create relation %u/%u/%u.%u",
-		 RelnGetSpcOid(reln),
-		 RelnGetDbOid(reln), RelnGetRelNumber(reln),
+		 reln->smgr_rnode.node.spcNode,
+		 reln->smgr_rnode.node.dbNode,
+		 reln->smgr_rnode.node.relNode,
 		 forkNum);
 
 	/*
@@ -1619,12 +1591,12 @@ neon_create(SMgrRelation reln, ForkNumber forkNum, bool isRedo)
 	 */
 	if (isRedo)
 	{
-		update_cached_relsize(rnode, forkNum, 0);
-		get_cached_relsize(rnode, forkNum,
+		update_cached_relsize(reln->smgr_rnode.node, forkNum, 0);
+		get_cached_relsize(reln->smgr_rnode.node, forkNum,
 						   &reln->smgr_cached_nblocks[forkNum]);
 	}
 	else
-		set_cached_relsize(rnode, forkNum, 0);
+		set_cached_relsize(reln->smgr_rnode.node, forkNum, 0);
 
 #ifdef DEBUG_COMPARE_LOCAL
 	if (IS_LOCAL_REL(reln))
@@ -1661,12 +1633,7 @@ neon_unlink(RelFileNodeBackend rnode, ForkNumber forkNum, bool isRedo)
 	mdunlink(rnode, forkNum, isRedo);
 	if (!RelFileNodeBackendIsTemp(rnode))
 	{
-
-#if PG_VERSION_NUM >= 160000
-		forget_cached_relsize(rnode.locator, forkNum);
-#else
 		forget_cached_relsize(rnode.node, forkNum);
-#endif
 	}
 }
 
@@ -1685,7 +1652,6 @@ neon_extend(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno,
 {
 	XLogRecPtr	lsn;
 	BlockNumber	n_blocks = 0;
-	RelFileNode rnode = RelnGetRnode(reln);
 
 	switch (reln->smgr_relpersistence)
 	{
@@ -1735,16 +1701,17 @@ neon_extend(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno,
 		neon_wallog_page(reln, forkNum, n_blocks++, buffer, true);
 
 	neon_wallog_page(reln, forkNum, blkno, buffer, false);
-	set_cached_relsize(rnode, forkNum, blkno + 1);
+	set_cached_relsize(reln->smgr_rnode.node, forkNum, blkno + 1);
 
 	lsn = PageGetLSN(buffer);
 	elog(SmgrTrace, "smgrextend called for %u/%u/%u.%u blk %u, page LSN: %X/%08X",
-		 RelnGetSpcOid(reln),
-		 RelnGetDbOid(reln), RelnGetRelNumber(reln),
+		 reln->smgr_rnode.node.spcNode,
+		 reln->smgr_rnode.node.dbNode,
+		 reln->smgr_rnode.node.relNode,
 		 forkNum, blkno,
 		 (uint32) (lsn >> 32), (uint32) lsn);
 
-	lfc_write(rnode, forkNum, blkno, buffer);
+	lfc_write(reln->smgr_rnode.node, forkNum, blkno, buffer);
 
 #ifdef DEBUG_COMPARE_LOCAL
 	if (IS_LOCAL_REL(reln))
@@ -1759,9 +1726,9 @@ neon_extend(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno,
 	if (lsn == InvalidXLogRecPtr)
 	{
 		lsn = GetXLogInsertRecPtr();
-		SetLastWrittenLSNForBlock(lsn, rnode, forkNum, blkno);
+		SetLastWrittenLSNForBlock(lsn, reln->smgr_rnode.node, forkNum, blkno);
 	}
-	SetLastWrittenLSNForRelation(lsn, rnode, forkNum);
+	SetLastWrittenLSNForRelation(lsn, reln->smgr_rnode.node, forkNum);
 }
 
 /*
@@ -1805,8 +1772,6 @@ neon_prefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum)
 	BufferTag	tag;
 	uint64		ring_index PG_USED_FOR_ASSERTS_ONLY;
 
-	RelFileNode rnode = RelnGetRnode(reln);
-
 	switch (reln->smgr_relpersistence)
 	{
 		case 0: /* probably shouldn't happen, but ignore it */
@@ -1821,18 +1786,15 @@ neon_prefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum)
 			elog(ERROR, "unknown relpersistence '%c'", reln->smgr_relpersistence);
 	}
 
-	if (lfc_cache_contains(rnode, forknum, blocknum))
+	if (lfc_cache_contains(reln->smgr_rnode.node, forknum, blocknum))
 		return false;
 
-#if PG_VERSION_NUM >= 160000
-	InitBufferTag(&tag, &rnode, forknum, blocknum);
-#else
 	tag = (BufferTag) {
-		.rnode = rnode,
+		.rnode = reln->smgr_rnode.node,
 		.forkNum = forknum,
 		.blockNum = blocknum
 	};
-#endif
+
 	ring_index = prefetch_register_buffer(tag, NULL, NULL);
 
 	Assert(ring_index < MyPState->ring_unused &&
@@ -1893,15 +1855,11 @@ neon_read_at_lsn(RelFileNode rnode, ForkNumber forkNum, BlockNumber blkno,
 	PrfHashEntry *entry;
 	PrefetchRequest *slot;
 
-#if PG_VERSION_NUM >= 160000
-	InitBufferTag(&buftag, &rnode, forkNum, blkno);
-#else
 	buftag = (BufferTag) {
 		.rnode = rnode,
 		.forkNum = forkNum,
-		.blockNum = blkno
+		.blockNum = blkno,
 	};
-#endif
 
 	/*
 	 * The redo process does not lock pages that it needs to replay but are
@@ -2001,9 +1959,9 @@ neon_read_at_lsn(RelFileNode rnode, ForkNumber forkNum, BlockNumber blkno,
 					(errcode(ERRCODE_IO_ERROR),
 					 errmsg("could not read block %u in rel %u/%u/%u.%u from page server at lsn %X/%08X",
 							blkno,
-							RnodeGetSpcOid(rnode),
-							RnodeGetDbOid(rnode),
-							RnodeGetRelNumber(rnode),
+							rnode.spcNode,
+							rnode.dbNode,
+							rnode.relNode,
 							forkNum,
 							(uint32) (request_lsn >> 32), (uint32) request_lsn),
 					 errdetail("page server returned error: %s",
@@ -2027,7 +1985,6 @@ neon_read(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno,
 {
 	bool		latest;
 	XLogRecPtr	request_lsn;
-	RelFileNode rnode = RelnGetRnode(reln);
 
 	switch (reln->smgr_relpersistence)
 	{
@@ -2047,13 +2004,13 @@ neon_read(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno,
 	}
 
 	/* Try to read from local file cache */
-	if (lfc_read(RelnGetRnode(reln), forkNum, blkno, buffer))
+	if (lfc_read(reln->smgr_rnode.node, forkNum, blkno, buffer))
 	{
 		return;
 	}
 
-	request_lsn = neon_get_request_lsn(&latest, rnode, forkNum, blkno);
-	neon_read_at_lsn(rnode, forkNum, blkno, request_lsn, latest, buffer);
+	request_lsn = neon_get_request_lsn(&latest, reln->smgr_rnode.node, forkNum, blkno);
+	neon_read_at_lsn(reln->smgr_rnode.node, forkNum, blkno, request_lsn, latest, buffer);
 
 #ifdef DEBUG_COMPARE_LOCAL
 	if (forkNum == MAIN_FORKNUM && IS_LOCAL_REL(reln))
@@ -2073,9 +2030,9 @@ neon_read(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno,
 			{
 				elog(PANIC, "page is new in MD but not in Page Server at blk %u in rel %u/%u/%u fork %u (request LSN %X/%08X):\n%s\n",
 					 blkno,
-					 RelnGetSpcOid(reln),
-					 RelnGetDbOid(reln),
-					 RelnGetRelNumber(reln),
+					 reln->smgr_rnode.node.spcNode,
+					 reln->smgr_rnode.node.dbNode,
+					 reln->smgr_rnode.node.relNode,
 					 forkNum,
 					 (uint32) (request_lsn >> 32), (uint32) request_lsn,
 					 hexdump_page(buffer));
@@ -2085,9 +2042,9 @@ neon_read(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno,
 		{
 			elog(PANIC, "page is new in Page Server but not in MD at blk %u in rel %u/%u/%u fork %u (request LSN %X/%08X):\n%s\n",
 				 blkno,
-				 RelnGetSpcOid(reln),
-				 RelnGetDbOid(reln),
-				 RelnGetRelNumber(reln),
+				 reln->smgr_rnode.node.spcNode,
+				 reln->smgr_rnode.node.dbNode,
+				 reln->smgr_rnode.node.relNode,
 				 forkNum,
 				 (uint32) (request_lsn >> 32), (uint32) request_lsn,
 				 hexdump_page(mdbuf));
@@ -2102,9 +2059,9 @@ neon_read(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno,
 			{
 				elog(PANIC, "heap buffers differ at blk %u in rel %u/%u/%u fork %u (request LSN %X/%08X):\n------ MD ------\n%s\n------ Page Server ------\n%s\n",
 					 blkno,
-					 RelnGetSpcOid(reln),
-					 RelnGetDbOid(reln),
-					 RelnGetRelNumber(reln),
+					 reln->smgr_rnode.node.spcNode,
+					 reln->smgr_rnode.node.dbNode,
+					 reln->smgr_rnode.node.relNode,
 					 forkNum,
 					 (uint32) (request_lsn >> 32), (uint32) request_lsn,
 					 hexdump_page(mdbuf_masked),
@@ -2123,9 +2080,9 @@ neon_read(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno,
 				{
 					elog(PANIC, "btree buffers differ at blk %u in rel %u/%u/%u fork %u (request LSN %X/%08X):\n------ MD ------\n%s\n------ Page Server ------\n%s\n",
 						 blkno,
-						 RelnGetSpcOid(reln),
-						 RelnGetDbOid(reln),
-						 RelnGetRelNumber(reln),
+						 reln->smgr_rnode.node.spcNode,
+						 reln->smgr_rnode.node.dbNode,
+						 reln->smgr_rnode.node.relNode,
 						 forkNum,
 						 (uint32) (request_lsn >> 32), (uint32) request_lsn,
 						 hexdump_page(mdbuf_masked),
@@ -2170,7 +2127,7 @@ neon_write(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 		   char *buffer, bool skipFsync)
 {
 	XLogRecPtr	lsn;
-	RelFileNode rnode = RelnGetRnode(reln);
+
 	switch (reln->smgr_relpersistence)
 	{
 		case 0:
@@ -2207,12 +2164,13 @@ neon_write(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 
 	lsn = PageGetLSN(buffer);
 	elog(SmgrTrace, "smgrwrite called for %u/%u/%u.%u blk %u, page LSN: %X/%08X",
-		 RelnGetSpcOid(reln),
-		 RelnGetDbOid(reln), RelnGetRelNumber(reln),
+		 reln->smgr_rnode.node.spcNode,
+		 reln->smgr_rnode.node.dbNode,
+		 reln->smgr_rnode.node.relNode,
 		 forknum, blocknum,
 		 (uint32) (lsn >> 32), (uint32) lsn);
 
-	lfc_write(rnode, forknum, blocknum, buffer);
+	lfc_write(reln->smgr_rnode.node, forknum, blocknum, buffer);
 
 #ifdef DEBUG_COMPARE_LOCAL
 	if (IS_LOCAL_REL(reln))
@@ -2230,7 +2188,6 @@ neon_nblocks(SMgrRelation reln, ForkNumber forknum)
 	BlockNumber n_blocks;
 	bool		latest;
 	XLogRecPtr	request_lsn;
-	RelFileNode rnode = RelnGetRnode(reln);
 
 	switch (reln->smgr_relpersistence)
 	{
@@ -2249,23 +2206,23 @@ neon_nblocks(SMgrRelation reln, ForkNumber forknum)
 			elog(ERROR, "unknown relpersistence '%c'", reln->smgr_relpersistence);
 	}
 
-	if (get_cached_relsize(RelnGetRnode(reln), forknum, &n_blocks))
+	if (get_cached_relsize(reln->smgr_rnode.node, forknum, &n_blocks))
 	{
 		elog(SmgrTrace, "cached nblocks for %u/%u/%u.%u: %u blocks",
-			 RelnGetSpcOid(reln),
-			 RelnGetDbOid(reln),
-			 RelnGetRelNumber(reln),
+			 reln->smgr_rnode.node.spcNode,
+			 reln->smgr_rnode.node.dbNode,
+			 reln->smgr_rnode.node.relNode,
 			 forknum, n_blocks);
 		return n_blocks;
 	}
 
-	request_lsn = neon_get_request_lsn(&latest, rnode, forknum, REL_METADATA_PSEUDO_BLOCKNO);
+	request_lsn = neon_get_request_lsn(&latest, reln->smgr_rnode.node, forknum, REL_METADATA_PSEUDO_BLOCKNO);
 	{
 		NeonNblocksRequest request = {
 			.req.tag = T_NeonNblocksRequest,
 			.req.latest = latest,
 			.req.lsn = request_lsn,
-			.rnode = rnode,
+			.rnode = reln->smgr_rnode.node,
 			.forknum = forknum,
 		};
 
@@ -2282,9 +2239,9 @@ neon_nblocks(SMgrRelation reln, ForkNumber forknum)
 			ereport(ERROR,
 					(errcode(ERRCODE_IO_ERROR),
 					 errmsg("could not read relation size of rel %u/%u/%u.%u from page server at lsn %X/%08X",
-							RelnGetSpcOid(reln),
-							RelnGetDbOid(reln),
-							RelnGetRelNumber(reln),
+							reln->smgr_rnode.node.spcNode,
+							reln->smgr_rnode.node.dbNode,
+							reln->smgr_rnode.node.relNode,
 							forknum,
 							(uint32) (request_lsn >> 32), (uint32) request_lsn),
 					 errdetail("page server returned error: %s",
@@ -2294,11 +2251,12 @@ neon_nblocks(SMgrRelation reln, ForkNumber forknum)
 		default:
 			elog(ERROR, "unexpected response from page server with tag 0x%02x", resp->tag);
 	}
-	update_cached_relsize(rnode, forknum, n_blocks);
+	update_cached_relsize(reln->smgr_rnode.node, forknum, n_blocks);
 
 	elog(SmgrTrace, "neon_nblocks: rel %u/%u/%u fork %u (request LSN %X/%08X): %u blocks",
-		 RelnGetSpcOid(reln),
-		 RelnGetDbOid(reln), RelnGetRelNumber(reln),
+		 reln->smgr_rnode.node.spcNode,
+		 reln->smgr_rnode.node.dbNode,
+		 reln->smgr_rnode.node.relNode,
 		 forknum,
 		 (uint32) (request_lsn >> 32), (uint32) request_lsn,
 		 n_blocks);
@@ -2311,7 +2269,7 @@ neon_nblocks(SMgrRelation reln, ForkNumber forknum)
  *	neon_db_size() -- Get the size of the database in bytes.
  */
 int64
-neon_dbsize(Oid dbOid)
+neon_dbsize(Oid dbNode)
 {
 	NeonResponse *resp;
 	int64		db_size;
@@ -2325,7 +2283,7 @@ neon_dbsize(Oid dbOid)
 			.req.tag = T_NeonDbSizeRequest,
 			.req.latest = latest,
 			.req.lsn = request_lsn,
-			.dbOid = dbOid,
+			.dbNode = dbNode,
 		};
 
 		resp = page_server_request(&request);
@@ -2341,7 +2299,7 @@ neon_dbsize(Oid dbOid)
 			ereport(ERROR,
 					(errcode(ERRCODE_IO_ERROR),
 					 errmsg("could not read db size of db %u from page server at lsn %X/%08X",
-							dbOid,
+							dbNode,
 							(uint32) (request_lsn >> 32), (uint32) request_lsn),
 					 errdetail("page server returned error: %s",
 							   ((NeonErrorResponse *) resp)->message)));
@@ -2352,7 +2310,7 @@ neon_dbsize(Oid dbOid)
 	}
 
 	elog(SmgrTrace, "neon_dbsize: db %u (request LSN %X/%08X): %ld bytes",
-		 dbOid,
+		 dbNode,
 		 (uint32) (request_lsn >> 32), (uint32) request_lsn,
 		 db_size);
 
@@ -2367,7 +2325,6 @@ void
 neon_truncate(SMgrRelation reln, ForkNumber forknum, BlockNumber nblocks)
 {
 	XLogRecPtr	lsn;
-	RelFileNode rnode = RelnGetRnode(reln);
 
 	switch (reln->smgr_relpersistence)
 	{
@@ -2387,7 +2344,7 @@ neon_truncate(SMgrRelation reln, ForkNumber forknum, BlockNumber nblocks)
 			elog(ERROR, "unknown relpersistence '%c'", reln->smgr_relpersistence);
 	}
 
-	set_cached_relsize(rnode, forknum, nblocks);
+	set_cached_relsize(reln->smgr_rnode.node, forknum, nblocks);
 
 	/*
 	 * Truncating a relation drops all its buffers from the buffer cache
@@ -2415,7 +2372,7 @@ neon_truncate(SMgrRelation reln, ForkNumber forknum, BlockNumber nblocks)
 	 * for the extended pages, so there's no harm in leaving behind obsolete
 	 * entries for the truncated chunks.
 	 */
-	SetLastWrittenLSNForRelation(lsn, rnode, forknum);
+	SetLastWrittenLSNForRelation(lsn, reln->smgr_rnode.node, forknum);
 
 #ifdef DEBUG_COMPARE_LOCAL
 	if (IS_LOCAL_REL(reln))
@@ -2485,9 +2442,9 @@ neon_start_unlogged_build(SMgrRelation reln)
 
 	ereport(SmgrTrace,
 			(errmsg("starting unlogged build of relation %u/%u/%u",
-					RelnGetSpcOid(reln),
-					RelnGetDbOid(reln),
-					RelnGetRelNumber(reln))));
+					reln->smgr_rnode.node.spcNode,
+					reln->smgr_rnode.node.dbNode,
+					reln->smgr_rnode.node.relNode)));
 
 	switch (reln->smgr_relpersistence)
 	{
@@ -2537,9 +2494,9 @@ neon_finish_unlogged_build_phase_1(SMgrRelation reln)
 
 	ereport(SmgrTrace,
 			(errmsg("finishing phase 1 of unlogged build of relation %u/%u/%u",
-					RelnGetSpcOid(reln),
-					RelnGetDbOid(reln),
-					RelnGetRelNumber(reln))));
+					reln->smgr_rnode.node.spcNode,
+					reln->smgr_rnode.node.dbNode,
+					reln->smgr_rnode.node.relNode)));
 
 	if (unlogged_build_phase == UNLOGGED_BUILD_NOT_PERMANENT)
 		return;
@@ -2566,9 +2523,9 @@ neon_end_unlogged_build(SMgrRelation reln)
 
 	ereport(SmgrTrace,
 			(errmsg("ending unlogged build of relation %u/%u/%u",
-					RelnGetSpcOid(reln),
-					RelnGetDbOid(reln),
-					RelnGetRelNumber(reln))));
+					reln->smgr_rnode.node.spcNode,
+					reln->smgr_rnode.node.dbNode,
+					reln->smgr_rnode.node.relNode)));
 
 	if (unlogged_build_phase != UNLOGGED_BUILD_NOT_PERMANENT)
 	{
@@ -2581,24 +2538,16 @@ neon_end_unlogged_build(SMgrRelation reln)
 		reln->smgr_relpersistence = RELPERSISTENCE_PERMANENT;
 
 		/* Remove local copy */
-#if PG_VERSION_NUM >= 160000
-		rnode.locator = RelnGetRnode(reln);
-#else
-		rnode.node = RelnGetRnode(reln);
-#endif
+		rnode = reln->smgr_rnode;
 		for (int forknum = 0; forknum <= MAX_FORKNUM; forknum++)
 		{
 			elog(SmgrTrace, "forgetting cached relsize for %u/%u/%u.%u",
-				 RelnGetSpcOid(reln),
-				 RelnGetDbOid(reln),
-				 RelnGetRelNumber(reln),
+				 rnode.node.spcNode,
+				 rnode.node.dbNode,
+				 rnode.node.relNode,
 				 forknum);
 
-#if PG_VERSION_NUM >= 160000
-			forget_cached_relsize(rnode.locator, forknum);
-#else
 			forget_cached_relsize(rnode.node, forknum);
-#endif
 			mdclose(reln, forknum);
 			/* use isRedo == true, so that we drop it immediately */
 			mdunlink(rnode, forknum, true);
@@ -2751,16 +2700,10 @@ neon_redo_read_buffer_filter(XLogReaderState *record, uint8 block_id)
 	 * regardless of whether the block is stored in shared buffers.
 	 * See also this function's top comment.
 	 */
-
-	if (!OidIsValid(RnodeGetDbOid(rnode)))
+	if (!OidIsValid(rnode.dbNode))
 		return false;
 
-#if PG_VERSION_NUM >= 160000
-	InitBufferTag(&tag, &rnode, forknum, blkno);
-#else
 	INIT_BUFFERTAG(tag, rnode, forknum, blkno);
-#endif
-
 	hash = BufTableHashCode(&tag);
 	partitionLock = BufMappingPartitionLock(hash);
 

pgxn/neon/relsize_cache.c

@@ -15,11 +15,7 @@
 #include "postgres.h"
 
 #include "pagestore_client.h"
-#if PG_VERSION_NUM >= 160000
-#include "storage/relfilelocator.h"
-#else
 #include "storage/relfilenode.h"
-#endif
 #include "storage/smgr.h"
 #include "storage/lwlock.h"
 #include "storage/ipc.h"
@@ -32,7 +28,6 @@
 #include "miscadmin.h"
 #endif
 
-
 typedef struct
 {
 	RelFileNode rnode;

pgxn/neon/walproposer.c

@@ -1394,12 +1394,7 @@ WalProposerRecovery(int donor, TimeLineID timeline, XLogRecPtr startpos, XLogRec
 	WalReceiverConn *wrconn;
 	WalRcvStreamOptions options;
 
-#if PG_VERSION_NUM >= 160000
-	bool must_use_password = false;
-	wrconn = walrcv_connect(safekeeper[donor].conninfo, false, must_use_password, "wal_proposer_recovery", &err);
-#else
 	wrconn = walrcv_connect(safekeeper[donor].conninfo, false, "wal_proposer_recovery", &err);
-#endif
 	if (!wrconn)
 	{
 		ereport(WARNING,
@@ -2236,18 +2231,6 @@ HandleSafekeeperResponse(void)
 		if (n_synced >= quorum)
 		{
 			/* All safekeepers synced! */
-			
-			/*
-			 * Send empty message to broadcast latest truncateLsn to all safekeepers.
-			 * This helps to finish next sync-safekeepers eailier, by skipping recovery
-			 * step.
-			 * 
-			 * We don't need to wait for response because it doesn't affect correctness,
-			 * and TCP should be able to deliver the message to safekeepers in case of
-			 * network working properly.
-			 */
-			BroadcastAppendRequest();
-
 			fprintf(stdout, "%X/%X\n", LSN_FORMAT_ARGS(propEpochStartLsn));
 			exit(0);
 		}