fix dbname

.github/workflows/benchmarking.yml

@@ -671,10 +671,6 @@ jobs:
         password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
       options: --init
 
-    # Increase timeout to 12h, default timeout is 6h
-    # we have regression in clickbench causing it to run 2-3x longer
-    timeout-minutes: 720
-
     steps:
     - uses: actions/checkout@v4
 
@@ -720,7 +716,7 @@ jobs:
         test_selection: performance/test_perf_olap.py
         run_in_parallel: false
         save_perf_report: ${{ env.SAVE_PERF_REPORT }}
-        extra_params: -m remote_cluster --timeout 43200 -k test_clickbench
+        extra_params: -m remote_cluster --timeout 21600 -k test_clickbench
         pg_version: ${{ env.DEFAULT_PG_VERSION }}
       env:
         VIP_VAP_ACCESS_TOKEN: "${{ secrets.VIP_VAP_ACCESS_TOKEN }}"

Cargo.lock

@@ -3749,7 +3749,6 @@ dependencies = [
  "tracing",
  "url",
  "utils",
- "wal_decoder",
  "walkdir",
  "workspace_hack",
 ]
@@ -4187,7 +4186,6 @@ dependencies = [
  "regex",
  "serde",
  "thiserror",
- "tracing",
  "utils",
 ]
 
@@ -6274,7 +6272,7 @@ dependencies = [
 [[package]]
 name = "tokio-epoll-uring"
 version = "0.1.0"
-source = "git+https://github.com/neondatabase/tokio-epoll-uring.git?branch=main#33e00106a268644d02ba0461bbd64476073b0ee1"
+source = "git+https://github.com/neondatabase/tokio-epoll-uring.git?branch=main#cb2dcea2058034bc209e7917b01c5097712a3168"
 dependencies = [
  "futures",
  "nix 0.26.4",
@@ -6790,7 +6788,7 @@ dependencies = [
 [[package]]
 name = "uring-common"
 version = "0.1.0"
-source = "git+https://github.com/neondatabase/tokio-epoll-uring.git?branch=main#33e00106a268644d02ba0461bbd64476073b0ee1"
+source = "git+https://github.com/neondatabase/tokio-epoll-uring.git?branch=main#cb2dcea2058034bc209e7917b01c5097712a3168"
 dependencies = [
  "bytes",
  "io-uring",
@@ -6956,20 +6954,6 @@ dependencies = [
  "utils",
 ]
 
-[[package]]
-name = "wal_decoder"
-version = "0.1.0"
-dependencies = [
- "anyhow",
- "bytes",
- "pageserver_api",
- "postgres_ffi",
- "serde",
- "tracing",
- "utils",
- "workspace_hack",
-]
-
 [[package]]
 name = "walkdir"
 version = "2.3.3"

Cargo.toml

@@ -33,7 +33,6 @@ members = [
     "libs/postgres_ffi/wal_craft",
     "libs/vm_monitor",
     "libs/walproposer",
-    "libs/wal_decoder",
 ]
 
 [workspace.package]
@@ -239,7 +238,6 @@ tracing-utils = { version = "0.1", path = "./libs/tracing-utils/" }
 utils = { version = "0.1", path = "./libs/utils/" }
 vm_monitor = { version = "0.1", path = "./libs/vm_monitor/" }
 walproposer = { version = "0.1", path = "./libs/walproposer/" }
-wal_decoder = { version = "0.1", path = "./libs/wal_decoder" }
 
 ## Common library dependency
 workspace_hack = { version = "0.1", path = "./workspace_hack/" }

compute/compute-node.Dockerfile

@@ -873,85 +873,6 @@ RUN curl -sSO https://static.rust-lang.org/rustup/dist/$(uname -m)-unknown-linux
 
 USER root
 
-#########################################################################################
-#
-# Layer "rust extensions pgrx12"
-#
-# pgrx started to support Postgres 17 since version 12,
-# but some older extension aren't compatible with it.
-# This layer should be used as a base for new pgrx extensions,
-# and eventually get merged with `rust-extensions-build`
-#
-#########################################################################################
-FROM build-deps AS rust-extensions-build-pgrx12
-ARG PG_VERSION
-COPY --from=pg-build /usr/local/pgsql/ /usr/local/pgsql/
-
-RUN apt-get update && \
-    apt-get install --no-install-recommends -y curl libclang-dev && \
-    useradd -ms /bin/bash nonroot -b /home
-
-ENV HOME=/home/nonroot
-ENV PATH="/home/nonroot/.cargo/bin:/usr/local/pgsql/bin/:$PATH"
-USER nonroot
-WORKDIR /home/nonroot
-
-RUN curl -sSO https://static.rust-lang.org/rustup/dist/$(uname -m)-unknown-linux-gnu/rustup-init && \
-    chmod +x rustup-init && \
-    ./rustup-init -y --no-modify-path --profile minimal --default-toolchain stable && \
-    rm rustup-init && \
-    cargo install --locked --version 0.12.6 cargo-pgrx && \
-    /bin/bash -c 'cargo pgrx init --pg${PG_VERSION:1}=/usr/local/pgsql/bin/pg_config'
-
-USER root
-
-#########################################################################################
-#
-# Layers "pg-onnx-build" and "pgrag-pg-build"
-# Compile "pgrag" extensions
-#
-#########################################################################################
-
-FROM rust-extensions-build-pgrx12 AS pg-onnx-build
-
-# cmake 3.26 or higher is required, so installing it using pip (bullseye-backports has cmake 3.25).
-# Install it using virtual environment, because Python 3.11 (the default version on Debian 12 (Bookworm)) complains otherwise
-RUN apt-get update && apt-get install -y python3 python3-pip python3-venv && \
-    python3 -m venv venv && \
-    . venv/bin/activate && \
-    python3 -m pip install cmake==3.30.5 && \
-    wget https://github.com/microsoft/onnxruntime/archive/refs/tags/v1.18.1.tar.gz -O onnxruntime.tar.gz && \
-    mkdir onnxruntime-src && cd onnxruntime-src && tar xzf ../onnxruntime.tar.gz --strip-components=1 -C . && \
-    ./build.sh --config Release --parallel --skip_submodule_sync --skip_tests --allow_running_as_root
-
-
-FROM pg-onnx-build AS pgrag-pg-build
-
-RUN apt-get install -y protobuf-compiler && \
-    wget https://github.com/neondatabase-labs/pgrag/archive/refs/tags/v0.0.0.tar.gz -O pgrag.tar.gz &&  \
-    echo "2cbe394c1e74fc8bcad9b52d5fbbfb783aef834ca3ce44626cfd770573700bb4 pgrag.tar.gz" | sha256sum --check && \
-    mkdir pgrag-src && cd pgrag-src && tar xzf ../pgrag.tar.gz --strip-components=1 -C . && \
-    \
-    cd exts/rag && \
-    sed -i 's/pgrx = "0.12.6"/pgrx = { version = "0.12.6", features = [ "unsafe-postgres" ] }/g' Cargo.toml && \
-    cargo pgrx install --release && \
-    echo "trusted = true" >> /usr/local/pgsql/share/extension/rag.control && \
-    \
-    cd ../rag_bge_small_en_v15 && \
-    sed -i 's/pgrx = "0.12.6"/pgrx = { version = "0.12.6", features = [ "unsafe-postgres" ] }/g' Cargo.toml && \
-    ORT_LIB_LOCATION=/home/nonroot/onnxruntime-src/build/Linux \
-        REMOTE_ONNX_URL=http://pg-ext-s3-gateway/pgrag-data/bge_small_en_v15.onnx \
-        cargo pgrx install --release --features remote_onnx && \
-    echo "trusted = true" >> /usr/local/pgsql/share/extension/rag_bge_small_en_v15.control && \
-    \
-    cd ../rag_jina_reranker_v1_tiny_en && \
-    sed -i 's/pgrx = "0.12.6"/pgrx = { version = "0.12.6", features = [ "unsafe-postgres" ] }/g' Cargo.toml && \
-    ORT_LIB_LOCATION=/home/nonroot/onnxruntime-src/build/Linux \
-        REMOTE_ONNX_URL=http://pg-ext-s3-gateway/pgrag-data/jina_reranker_v1_tiny_en.onnx \
-        cargo pgrx install --release --features remote_onnx && \
-    echo "trusted = true" >> /usr/local/pgsql/share/extension/rag_jina_reranker_v1_tiny_en.control
-
-
 #########################################################################################
 #
 # Layer "pg-jsonschema-pg-build"
@@ -1164,7 +1085,6 @@ COPY --from=h3-pg-build /h3/usr /
 COPY --from=unit-pg-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=vector-pg-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pgjwt-pg-build /usr/local/pgsql/ /usr/local/pgsql/
-COPY --from=pgrag-pg-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg-jsonschema-pg-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg-graphql-pg-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg-tiktoken-pg-build /usr/local/pgsql/ /usr/local/pgsql/
@@ -1354,7 +1274,6 @@ COPY --from=unit-pg-build /postgresql-unit.tar.gz /ext-src/
 COPY --from=vector-pg-build /pgvector.tar.gz /ext-src/
 COPY --from=vector-pg-build /pgvector.patch /ext-src/
 COPY --from=pgjwt-pg-build /pgjwt.tar.gz /ext-src
-#COPY --from=pgrag-pg-build /usr/local/pgsql/ /usr/local/pgsql/
 #COPY --from=pg-jsonschema-pg-build /home/nonroot/pg_jsonschema.tar.gz /ext-src
 #COPY --from=pg-graphql-pg-build /home/nonroot/pg_graphql.tar.gz /ext-src
 #COPY --from=pg-tiktoken-pg-build /home/nonroot/pg_tiktoken.tar.gz /ext-src

control_plane/src/pageserver.rs

@@ -17,7 +17,7 @@ use std::time::Duration;
 
 use anyhow::{bail, Context};
 use camino::Utf8PathBuf;
-use pageserver_api::models::{self, TenantInfo, TimelineInfo};
+use pageserver_api::models::{self, AuxFilePolicy, TenantInfo, TimelineInfo};
 use pageserver_api::shard::TenantShardId;
 use pageserver_client::mgmt_api;
 use postgres_backend::AuthType;
@@ -399,6 +399,11 @@ impl PageServerNode {
                 .map(serde_json::from_str)
                 .transpose()
                 .context("parse `timeline_get_throttle` from json")?,
+            switch_aux_file_policy: settings
+                .remove("switch_aux_file_policy")
+                .map(|x| x.parse::<AuxFilePolicy>())
+                .transpose()
+                .context("Failed to parse 'switch_aux_file_policy'")?,
             lsn_lease_length: settings.remove("lsn_lease_length").map(|x| x.to_string()),
             lsn_lease_length_for_ts: settings
                 .remove("lsn_lease_length_for_ts")
@@ -494,6 +499,11 @@ impl PageServerNode {
                     .map(serde_json::from_str)
                     .transpose()
                     .context("parse `timeline_get_throttle` from json")?,
+                switch_aux_file_policy: settings
+                    .remove("switch_aux_file_policy")
+                    .map(|x| x.parse::<AuxFilePolicy>())
+                    .transpose()
+                    .context("Failed to parse 'switch_aux_file_policy'")?,
                 lsn_lease_length: settings.remove("lsn_lease_length").map(|x| x.to_string()),
                 lsn_lease_length_for_ts: settings
                     .remove("lsn_lease_length_for_ts")

libs/pageserver_api/src/config.rs

@@ -250,6 +250,12 @@ pub struct TenantConfigToml {
     // Expresed in multiples of checkpoint distance.
     pub image_layer_creation_check_threshold: u8,
 
+    /// Switch to a new aux file policy. Switching this flag requires the user has not written any aux file into
+    /// the storage before, and this flag cannot be switched back. Otherwise there will be data corruptions.
+    /// There is a `last_aux_file_policy` flag which gets persisted in `index_part.json` once the first aux
+    /// file is written.
+    pub switch_aux_file_policy: crate::models::AuxFilePolicy,
+
     /// The length for an explicit LSN lease request.
     /// Layers needed to reconstruct pages at LSN will not be GC-ed during this interval.
     #[serde(with = "humantime_serde")]
@@ -469,6 +475,7 @@ impl Default for TenantConfigToml {
             lazy_slru_download: false,
             timeline_get_throttle: crate::models::ThrottleConfig::disabled(),
             image_layer_creation_check_threshold: DEFAULT_IMAGE_LAYER_CREATION_CHECK_THRESHOLD,
+            switch_aux_file_policy: crate::models::AuxFilePolicy::default_tenant_config(),
             lsn_lease_length: LsnLease::DEFAULT_LENGTH,
             lsn_lease_length_for_ts: LsnLease::DEFAULT_LENGTH_FOR_TS,
         }

libs/pageserver_api/src/lib.rs

@@ -5,11 +5,9 @@ pub mod controller_api;
 pub mod key;
 pub mod keyspace;
 pub mod models;
-pub mod record;
 pub mod reltag;
 pub mod shard;
 /// Public API types
 pub mod upcall_api;
-pub mod value;
 
 pub mod config;

libs/pageserver_api/src/models.rs

@@ -10,6 +10,7 @@ use std::{
     io::{BufRead, Read},
     num::{NonZeroU32, NonZeroU64, NonZeroUsize},
     str::FromStr,
+    sync::atomic::AtomicUsize,
     time::{Duration, SystemTime},
 };
 
@@ -308,6 +309,7 @@ pub struct TenantConfig {
     pub lazy_slru_download: Option<bool>,
     pub timeline_get_throttle: Option<ThrottleConfig>,
     pub image_layer_creation_check_threshold: Option<u8>,
+    pub switch_aux_file_policy: Option<AuxFilePolicy>,
     pub lsn_lease_length: Option<String>,
     pub lsn_lease_length_for_ts: Option<String>,
 }
@@ -348,6 +350,68 @@ pub enum AuxFilePolicy {
     CrossValidation,
 }
 
+impl AuxFilePolicy {
+    pub fn is_valid_migration_path(from: Option<Self>, to: Self) -> bool {
+        matches!(
+            (from, to),
+            (None, _) | (Some(AuxFilePolicy::CrossValidation), AuxFilePolicy::V2)
+        )
+    }
+
+    /// If a tenant writes aux files without setting `switch_aux_policy`, this value will be used.
+    pub fn default_tenant_config() -> Self {
+        Self::V2
+    }
+}
+
+/// The aux file policy memory flag. Users can store `Option<AuxFilePolicy>` into this atomic flag. 0 == unspecified.
+pub struct AtomicAuxFilePolicy(AtomicUsize);
+
+impl AtomicAuxFilePolicy {
+    pub fn new(policy: Option<AuxFilePolicy>) -> Self {
+        Self(AtomicUsize::new(
+            policy.map(AuxFilePolicy::to_usize).unwrap_or_default(),
+        ))
+    }
+
+    pub fn load(&self) -> Option<AuxFilePolicy> {
+        match self.0.load(std::sync::atomic::Ordering::Acquire) {
+            0 => None,
+            other => Some(AuxFilePolicy::from_usize(other)),
+        }
+    }
+
+    pub fn store(&self, policy: Option<AuxFilePolicy>) {
+        self.0.store(
+            policy.map(AuxFilePolicy::to_usize).unwrap_or_default(),
+            std::sync::atomic::Ordering::Release,
+        );
+    }
+}
+
+impl AuxFilePolicy {
+    pub fn to_usize(self) -> usize {
+        match self {
+            Self::V1 => 1,
+            Self::CrossValidation => 2,
+            Self::V2 => 3,
+        }
+    }
+
+    pub fn try_from_usize(this: usize) -> Option<Self> {
+        match this {
+            1 => Some(Self::V1),
+            2 => Some(Self::CrossValidation),
+            3 => Some(Self::V2),
+            _ => None,
+        }
+    }
+
+    pub fn from_usize(this: usize) -> Self {
+        Self::try_from_usize(this).unwrap()
+    }
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(tag = "kind")]
 pub enum EvictionPolicy {
@@ -1569,6 +1633,71 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_aux_file_migration_path() {
+        assert!(AuxFilePolicy::is_valid_migration_path(
+            None,
+            AuxFilePolicy::V1
+        ));
+        assert!(AuxFilePolicy::is_valid_migration_path(
+            None,
+            AuxFilePolicy::V2
+        ));
+        assert!(AuxFilePolicy::is_valid_migration_path(
+            None,
+            AuxFilePolicy::CrossValidation
+        ));
+        // Self-migration is not a valid migration path, and the caller should handle it by itself.
+        assert!(!AuxFilePolicy::is_valid_migration_path(
+            Some(AuxFilePolicy::V1),
+            AuxFilePolicy::V1
+        ));
+        assert!(!AuxFilePolicy::is_valid_migration_path(
+            Some(AuxFilePolicy::V2),
+            AuxFilePolicy::V2
+        ));
+        assert!(!AuxFilePolicy::is_valid_migration_path(
+            Some(AuxFilePolicy::CrossValidation),
+            AuxFilePolicy::CrossValidation
+        ));
+        // Migrations not allowed
+        assert!(!AuxFilePolicy::is_valid_migration_path(
+            Some(AuxFilePolicy::CrossValidation),
+            AuxFilePolicy::V1
+        ));
+        assert!(!AuxFilePolicy::is_valid_migration_path(
+            Some(AuxFilePolicy::V1),
+            AuxFilePolicy::V2
+        ));
+        assert!(!AuxFilePolicy::is_valid_migration_path(
+            Some(AuxFilePolicy::V2),
+            AuxFilePolicy::V1
+        ));
+        assert!(!AuxFilePolicy::is_valid_migration_path(
+            Some(AuxFilePolicy::V2),
+            AuxFilePolicy::CrossValidation
+        ));
+        assert!(!AuxFilePolicy::is_valid_migration_path(
+            Some(AuxFilePolicy::V1),
+            AuxFilePolicy::CrossValidation
+        ));
+        // Migrations allowed
+        assert!(AuxFilePolicy::is_valid_migration_path(
+            Some(AuxFilePolicy::CrossValidation),
+            AuxFilePolicy::V2
+        ));
+    }
+
+    #[test]
+    fn test_aux_parse() {
+        assert_eq!(AuxFilePolicy::from_str("V2").unwrap(), AuxFilePolicy::V2);
+        assert_eq!(AuxFilePolicy::from_str("v2").unwrap(), AuxFilePolicy::V2);
+        assert_eq!(
+            AuxFilePolicy::from_str("cross-validation").unwrap(),
+            AuxFilePolicy::CrossValidation
+        );
+    }
+
     #[test]
     fn test_image_compression_algorithm_parsing() {
         use ImageCompressionAlgorithm::*;

libs/pageserver_api/src/record.rs

@@ -1,113 +0,0 @@
-//! This module defines the WAL record format used within the pageserver.
-
-use bytes::Bytes;
-use postgres_ffi::walrecord::{describe_postgres_wal_record, MultiXactMember};
-use postgres_ffi::{MultiXactId, MultiXactOffset, TimestampTz, TransactionId};
-use serde::{Deserialize, Serialize};
-use utils::bin_ser::DeserializeError;
-
-/// Each update to a page is represented by a NeonWalRecord. It can be a wrapper
-/// around a PostgreSQL WAL record, or a custom neon-specific "record".
-#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
-pub enum NeonWalRecord {
-    /// Native PostgreSQL WAL record
-    Postgres { will_init: bool, rec: Bytes },
-
-    /// Clear bits in heap visibility map. ('flags' is bitmap of bits to clear)
-    ClearVisibilityMapFlags {
-        new_heap_blkno: Option<u32>,
-        old_heap_blkno: Option<u32>,
-        flags: u8,
-    },
-    /// Mark transaction IDs as committed on a CLOG page
-    ClogSetCommitted {
-        xids: Vec<TransactionId>,
-        timestamp: TimestampTz,
-    },
-    /// Mark transaction IDs as aborted on a CLOG page
-    ClogSetAborted { xids: Vec<TransactionId> },
-    /// Extend multixact offsets SLRU
-    MultixactOffsetCreate {
-        mid: MultiXactId,
-        moff: MultiXactOffset,
-    },
-    /// Extend multixact members SLRU.
-    MultixactMembersCreate {
-        moff: MultiXactOffset,
-        members: Vec<MultiXactMember>,
-    },
-    /// Update the map of AUX files, either writing or dropping an entry
-    AuxFile {
-        file_path: String,
-        content: Option<Bytes>,
-    },
-
-    /// A testing record for unit testing purposes. It supports append data to an existing image, or clear it.
-    #[cfg(feature = "testing")]
-    Test {
-        /// Append a string to the image.
-        append: String,
-        /// Clear the image before appending.
-        clear: bool,
-        /// Treat this record as an init record. `clear` should be set to true if this field is set
-        /// to true. This record does not need the history WALs to reconstruct. See [`NeonWalRecord::will_init`] and
-        /// its references in `timeline.rs`.
-        will_init: bool,
-    },
-}
-
-impl NeonWalRecord {
-    /// Does replaying this WAL record initialize the page from scratch, or does
-    /// it need to be applied over the previous image of the page?
-    pub fn will_init(&self) -> bool {
-        // If you change this function, you'll also need to change ValueBytes::will_init
-        match self {
-            NeonWalRecord::Postgres { will_init, rec: _ } => *will_init,
-            #[cfg(feature = "testing")]
-            NeonWalRecord::Test { will_init, .. } => *will_init,
-            // None of the special neon record types currently initialize the page
-            _ => false,
-        }
-    }
-
-    #[cfg(feature = "testing")]
-    pub fn wal_append(s: impl AsRef<str>) -> Self {
-        Self::Test {
-            append: s.as_ref().to_string(),
-            clear: false,
-            will_init: false,
-        }
-    }
-
-    #[cfg(feature = "testing")]
-    pub fn wal_clear() -> Self {
-        Self::Test {
-            append: "".to_string(),
-            clear: true,
-            will_init: false,
-        }
-    }
-
-    #[cfg(feature = "testing")]
-    pub fn wal_init() -> Self {
-        Self::Test {
-            append: "".to_string(),
-            clear: true,
-            will_init: true,
-        }
-    }
-}
-
-/// Build a human-readable string to describe a WAL record
-///
-/// For debugging purposes
-pub fn describe_wal_record(rec: &NeonWalRecord) -> Result<String, DeserializeError> {
-    match rec {
-        NeonWalRecord::Postgres { will_init, rec } => Ok(format!(
-            "will_init: {}, {}",
-            will_init,
-            describe_postgres_wal_record(rec)?
-        )),
-        _ => Ok(format!("{:?}", rec)),
-    }
-}

libs/postgres_ffi/Cargo.toml

@@ -15,7 +15,6 @@ memoffset.workspace = true
 thiserror.workspace = true
 serde.workspace = true
 utils.workspace = true
-tracing.workspace = true
 
 [dev-dependencies]
 env_logger.workspace = true

libs/postgres_ffi/src/lib.rs

@@ -217,7 +217,6 @@ macro_rules! enum_pgversion {
 
 pub mod pg_constants;
 pub mod relfile_utils;
-pub mod walrecord;
 
 // Export some widely used datatypes that are unlikely to change across Postgres versions
 pub use v14::bindings::RepOriginId;

libs/wal_decoder/Cargo.toml

@@ -1,18 +0,0 @@
-[package]
-name = "wal_decoder"
-version = "0.1.0"
-edition.workspace = true
-license.workspace = true
-
-[features]
-testing = []
-
-[dependencies]
-anyhow.workspace = true
-bytes.workspace = true
-pageserver_api.workspace = true
-postgres_ffi.workspace = true
-serde.workspace = true
-tracing.workspace = true
-utils.workspace = true
-workspace_hack = { version = "0.1", path = "../../workspace_hack" }

libs/wal_decoder/src/decoder.rs

@@ -1 +0,0 @@
-

libs/wal_decoder/src/lib.rs

@@ -1,2 +0,0 @@
-pub mod decoder;
-pub mod models;

libs/wal_decoder/src/models.rs

@@ -1,167 +0,0 @@
-//! This module houses types which represent decoded PG WAL records
-//! ready for the pageserver to interpret. They are derived from the original
-//! WAL records, so that each struct corresponds closely to one WAL record of
-//! a specific kind. They contain the same information as the original WAL records,
-//! just decoded into structs and fields for easier access.
-//!
-//! The ingestion code uses these structs to help with parsing the WAL records,
-//! and it splits them into a stream of modifications to the key-value pairs that
-//! are ultimately stored in delta layers.  See also the split-out counterparts in
-//! [`postgres_ffi::walrecord`].
-//!
-//! The pipeline which processes WAL records is not super obvious, so let's follow
-//! the flow of an example XACT_COMMIT Postgres record:
-//!
-//! (Postgres XACT_COMMIT record)
-//! |
-//! |--> pageserver::walingest::WalIngest::decode_xact_record
-//!      |
-//!      |--> ([`XactRecord::Commit`])
-//!           |
-//!           |--> pageserver::walingest::WalIngest::ingest_xact_record
-//!                |
-//!                |--> (NeonWalRecord::ClogSetCommitted)
-//!                     |
-//!                     |--> write to KV store within the pageserver
-
-use bytes::Bytes;
-use pageserver_api::reltag::{RelTag, SlruKind};
-use postgres_ffi::walrecord::{
-    XlMultiXactCreate, XlMultiXactTruncate, XlRelmapUpdate, XlReploriginDrop, XlReploriginSet,
-    XlSmgrTruncate, XlXactParsedRecord,
-};
-use postgres_ffi::{Oid, TransactionId};
-use utils::lsn::Lsn;
-
-pub enum HeapamRecord {
-    ClearVmBits(ClearVmBits),
-}
-
-pub struct ClearVmBits {
-    pub new_heap_blkno: Option<u32>,
-    pub old_heap_blkno: Option<u32>,
-    pub vm_rel: RelTag,
-    pub flags: u8,
-}
-
-pub enum NeonrmgrRecord {
-    ClearVmBits(ClearVmBits),
-}
-
-pub enum SmgrRecord {
-    Create(SmgrCreate),
-    Truncate(XlSmgrTruncate),
-}
-
-pub struct SmgrCreate {
-    pub rel: RelTag,
-}
-
-pub enum DbaseRecord {
-    Create(DbaseCreate),
-    Drop(DbaseDrop),
-}
-
-pub struct DbaseCreate {
-    pub db_id: Oid,
-    pub tablespace_id: Oid,
-    pub src_db_id: Oid,
-    pub src_tablespace_id: Oid,
-}
-
-pub struct DbaseDrop {
-    pub db_id: Oid,
-    pub tablespace_ids: Vec<Oid>,
-}
-
-pub enum ClogRecord {
-    ZeroPage(ClogZeroPage),
-    Truncate(ClogTruncate),
-}
-
-pub struct ClogZeroPage {
-    pub segno: u32,
-    pub rpageno: u32,
-}
-
-pub struct ClogTruncate {
-    pub pageno: u32,
-    pub oldest_xid: TransactionId,
-    pub oldest_xid_db: Oid,
-}
-
-pub enum XactRecord {
-    Commit(XactCommon),
-    Abort(XactCommon),
-    CommitPrepared(XactCommon),
-    AbortPrepared(XactCommon),
-    Prepare(XactPrepare),
-}
-
-pub struct XactCommon {
-    pub parsed: XlXactParsedRecord,
-    pub origin_id: u16,
-    // Fields below are only used for logging
-    pub xl_xid: TransactionId,
-    pub lsn: Lsn,
-}
-
-pub struct XactPrepare {
-    pub xl_xid: TransactionId,
-    pub data: Bytes,
-}
-
-pub enum MultiXactRecord {
-    ZeroPage(MultiXactZeroPage),
-    Create(XlMultiXactCreate),
-    Truncate(XlMultiXactTruncate),
-}
-
-pub struct MultiXactZeroPage {
-    pub slru_kind: SlruKind,
-    pub segno: u32,
-    pub rpageno: u32,
-}
-
-pub enum RelmapRecord {
-    Update(RelmapUpdate),
-}
-
-pub struct RelmapUpdate {
-    pub update: XlRelmapUpdate,
-    pub buf: Bytes,
-}
-
-pub enum XlogRecord {
-    Raw(RawXlogRecord),
-}
-
-pub struct RawXlogRecord {
-    pub info: u8,
-    pub lsn: Lsn,
-    pub buf: Bytes,
-}
-
-pub enum LogicalMessageRecord {
-    Put(PutLogicalMessage),
-    #[cfg(feature = "testing")]
-    Failpoint,
-}
-
-pub struct PutLogicalMessage {
-    pub path: String,
-    pub buf: Bytes,
-}
-
-pub enum StandbyRecord {
-    RunningXacts(StandbyRunningXacts),
-}
-
-pub struct StandbyRunningXacts {
-    pub oldest_running_xid: TransactionId,
-}
-
-pub enum ReploriginRecord {
-    Set(XlReploriginSet),
-    Drop(XlReploriginDrop),
-}

pageserver/Cargo.toml

@@ -8,7 +8,7 @@ license.workspace = true
 default = []
 # Enables test-only APIs, incuding failpoints. In particular, enables the `fail_point!` macro,
 # which adds some runtime cost to run tests on outage conditions
-testing = ["fail/failpoints", "pageserver_api/testing", "wal_decoder/testing"]
+testing = ["fail/failpoints", "pageserver_api/testing" ]
 
 [dependencies]
 anyhow.workspace = true
@@ -83,7 +83,6 @@ enum-map.workspace = true
 enumset = { workspace = true, features = ["serde"]}
 strum.workspace = true
 strum_macros.workspace = true
-wal_decoder.workspace = true
 
 [target.'cfg(target_os = "linux")'.dependencies]
 procfs.workspace = true

pageserver/benches/bench_ingest.rs

@@ -8,12 +8,13 @@ use pageserver::{
     context::{DownloadBehavior, RequestContext},
     l0_flush::{L0FlushConfig, L0FlushGlobalState},
     page_cache,
+    repository::Value,
     task_mgr::TaskKind,
     tenant::storage_layer::inmemory_layer::SerializedBatch,
     tenant::storage_layer::InMemoryLayer,
     virtual_file,
 };
-use pageserver_api::{key::Key, shard::TenantShardId, value::Value};
+use pageserver_api::{key::Key, shard::TenantShardId};
 use utils::{
     bin_ser::BeSer,
     id::{TenantId, TimelineId},

pageserver/benches/bench_layer_map.rs

@@ -1,9 +1,9 @@
 use criterion::measurement::WallTime;
 use pageserver::keyspace::{KeyPartitioning, KeySpace};
+use pageserver::repository::Key;
 use pageserver::tenant::layer_map::LayerMap;
 use pageserver::tenant::storage_layer::LayerName;
 use pageserver::tenant::storage_layer::PersistentLayerDesc;
-use pageserver_api::key::Key;
 use pageserver_api::shard::TenantShardId;
 use rand::prelude::{SeedableRng, SliceRandom, StdRng};
 use std::cmp::{max, min};

pageserver/benches/bench_walredo.rs

@@ -60,8 +60,7 @@ use anyhow::Context;
 use bytes::{Buf, Bytes};
 use criterion::{BenchmarkId, Criterion};
 use once_cell::sync::Lazy;
-use pageserver::{config::PageServerConf, walredo::PostgresRedoManager};
-use pageserver_api::record::NeonWalRecord;
+use pageserver::{config::PageServerConf, walrecord::NeonWalRecord, walredo::PostgresRedoManager};
 use pageserver_api::{key::Key, shard::TenantShardId};
 use std::{
     future::Future,

pageserver/ctl/src/draw_timeline_dir.rs

@@ -51,7 +51,7 @@
 //!
 
 use anyhow::{Context, Result};
-use pageserver_api::key::Key;
+use pageserver::repository::Key;
 use std::cmp::Ordering;
 use std::io::{self, BufRead};
 use std::path::PathBuf;

pageserver/ctl/src/layer_map_analyzer.rs

@@ -14,12 +14,12 @@ use std::ops::Range;
 use std::{fs, str};
 
 use pageserver::page_cache::{self, PAGE_SZ};
+use pageserver::repository::{Key, KEY_SIZE};
 use pageserver::tenant::block_io::FileBlockReader;
 use pageserver::tenant::disk_btree::{DiskBtreeReader, VisitDirection};
 use pageserver::tenant::storage_layer::delta_layer::{Summary, DELTA_KEY_SIZE};
 use pageserver::tenant::storage_layer::range_overlaps;
 use pageserver::virtual_file::{self, VirtualFile};
-use pageserver_api::key::{Key, KEY_SIZE};
 
 use utils::{bin_ser::BeSer, lsn::Lsn};
 

pageserver/ctl/src/layers.rs

@@ -14,13 +14,13 @@ use pageserver::tenant::{TENANTS_SEGMENT_NAME, TIMELINES_SEGMENT_NAME};
 use pageserver::virtual_file::api::IoMode;
 use pageserver::{page_cache, virtual_file};
 use pageserver::{
+    repository::{Key, KEY_SIZE},
     tenant::{
         block_io::FileBlockReader, disk_btree::VisitDirection,
         storage_layer::delta_layer::DELTA_KEY_SIZE,
     },
     virtual_file::VirtualFile,
 };
-use pageserver_api::key::{Key, KEY_SIZE};
 use std::fs;
 use utils::bin_ser::BeSer;
 use utils::id::{TenantId, TimelineId};

pageserver/pagebench/src/cmd/aux_files.rs

@@ -1,4 +1,4 @@
-use pageserver_api::models::{TenantConfig, TenantConfigRequest};
+use pageserver_api::models::{AuxFilePolicy, TenantConfig, TenantConfigRequest};
 use pageserver_api::shard::TenantShardId;
 use utils::id::TenantTimelineId;
 use utils::lsn::Lsn;
@@ -66,7 +66,10 @@ async fn main_impl(args: Args) -> anyhow::Result<()> {
     mgmt_api_client
         .tenant_config(&TenantConfigRequest {
             tenant_id: timeline.tenant_id,
-            config: TenantConfig::default(),
+            config: TenantConfig {
+                switch_aux_file_policy: Some(AuxFilePolicy::V2),
+                ..Default::default()
+            },
         })
         .await?;
 

pageserver/src/deletion_queue.rs

@@ -696,7 +696,7 @@ impl DeletionQueue {
 mod test {
     use camino::Utf8Path;
     use hex_literal::hex;
-    use pageserver_api::{key::Key, shard::ShardIndex, upcall_api::ReAttachResponseTenant};
+    use pageserver_api::{shard::ShardIndex, upcall_api::ReAttachResponseTenant};
     use std::{io::ErrorKind, time::Duration};
     use tracing::info;
 
@@ -705,6 +705,7 @@ mod test {
 
     use crate::{
         controller_upcall_client::RetryForeverError,
+        repository::Key,
         tenant::{harness::TenantHarness, storage_layer::DeltaLayerName},
     };
 

pageserver/src/http/routes.rs

@@ -2232,13 +2232,13 @@ async fn getpage_at_lsn_handler(
     check_permission(&request, Some(tenant_shard_id.tenant_id))?;
     let state = get_state(&request);
 
-    struct Key(pageserver_api::key::Key);
+    struct Key(crate::repository::Key);
 
     impl std::str::FromStr for Key {
         type Err = anyhow::Error;
 
         fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
-            pageserver_api::key::Key::from_hex(s).map(Key)
+            crate::repository::Key::from_hex(s).map(Key)
         }
     }
 

pageserver/src/import_datadir.rs

@@ -19,11 +19,12 @@ use crate::metrics::WAL_INGEST;
 use crate::pgdatadir_mapping::*;
 use crate::tenant::Timeline;
 use crate::walingest::WalIngest;
+use crate::walrecord::decode_wal_record;
+use crate::walrecord::DecodedWALRecord;
 use pageserver_api::reltag::{RelTag, SlruKind};
 use postgres_ffi::pg_constants;
 use postgres_ffi::relfile_utils::*;
 use postgres_ffi::waldecoder::WalStreamDecoder;
-use postgres_ffi::walrecord::{decode_wal_record, DecodedWALRecord};
 use postgres_ffi::ControlFileData;
 use postgres_ffi::DBState_DB_SHUTDOWNED;
 use postgres_ffi::Oid;

pageserver/src/lib.rs

@@ -24,6 +24,7 @@ pub mod metrics;
 pub mod page_cache;
 pub mod page_service;
 pub mod pgdatadir_mapping;
+pub mod repository;
 pub mod span;
 pub(crate) mod statvfs;
 pub mod task_mgr;
@@ -31,6 +32,7 @@ pub mod tenant;
 pub mod utilization;
 pub mod virtual_file;
 pub mod walingest;
+pub mod walrecord;
 pub mod walredo;
 
 use camino::Utf8Path;

pageserver/src/pgdatadir_mapping.rs

@@ -7,14 +7,14 @@
 //! Clarify that)
 //!
 use super::tenant::{PageReconstructError, Timeline};
-use crate::aux_file;
 use crate::context::RequestContext;
 use crate::keyspace::{KeySpace, KeySpaceAccum};
 use crate::span::debug_assert_current_span_has_tenant_and_timeline_id_no_shard_id;
+use crate::walrecord::NeonWalRecord;
+use crate::{aux_file, repository::*};
 use anyhow::{ensure, Context};
 use bytes::{Buf, Bytes, BytesMut};
 use enum_map::Enum;
-use pageserver_api::key::Key;
 use pageserver_api::key::{
     dbdir_key_range, rel_block_to_key, rel_dir_to_key, rel_key_range, rel_size_to_key,
     relmap_file_key, repl_origin_key, repl_origin_key_range, slru_block_to_key, slru_dir_to_key,
@@ -22,9 +22,7 @@ use pageserver_api::key::{
     CompactKey, AUX_FILES_KEY, CHECKPOINT_KEY, CONTROLFILE_KEY, DBDIR_KEY, TWOPHASEDIR_KEY,
 };
 use pageserver_api::keyspace::SparseKeySpace;
-use pageserver_api::record::NeonWalRecord;
 use pageserver_api::reltag::{BlockNumber, RelTag, SlruKind};
-use pageserver_api::value::Value;
 use postgres_ffi::relfile_utils::{FSM_FORKNUM, VISIBILITYMAP_FORKNUM};
 use postgres_ffi::BLCKSZ;
 use postgres_ffi::{Oid, RepOriginId, TimestampTz, TransactionId};

pageserver/src/repository.rs

@@ -1,16 +1,13 @@
-//! This module defines the value type used by the storage engine.
-//!
-//! A [`Value`] represents either a completely new value for one Key ([`Value::Image`]),
-//! or a "delta" of how to get from previous version of the value to the new one
-//! ([`Value::WalRecord`]])
-//!
-//! Note that the [`Value`] type is used for the permananent storage format, so any
-//! changes to it must be backwards compatible.
-
-use crate::record::NeonWalRecord;
+use crate::walrecord::NeonWalRecord;
+use anyhow::Result;
 use bytes::Bytes;
 use serde::{Deserialize, Serialize};
+use std::ops::AddAssign;
+use std::time::Duration;
 
+pub use pageserver_api::key::{Key, KEY_SIZE};
+
+/// A 'value' stored for a one Key.
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 pub enum Value {
     /// An Image value contains a full copy of the value
@@ -23,12 +20,10 @@ pub enum Value {
 }
 
 impl Value {
-    #[inline(always)]
     pub fn is_image(&self) -> bool {
         matches!(self, Value::Image(_))
     }
 
-    #[inline(always)]
     pub fn will_init(&self) -> bool {
         match self {
             Value::Image(_) => true,
@@ -38,18 +33,17 @@ impl Value {
 }
 
 #[derive(Debug, PartialEq)]
-pub enum InvalidInput {
+pub(crate) enum InvalidInput {
     TooShortValue,
     TooShortPostgresRecord,
 }
 
 /// We could have a ValueRef where everything is `serde(borrow)`. Before implementing that, lets
 /// use this type for querying if a slice looks some particular way.
-pub struct ValueBytes;
+pub(crate) struct ValueBytes;
 
 impl ValueBytes {
-    #[inline(always)]
-    pub fn will_init(raw: &[u8]) -> Result<bool, InvalidInput> {
+    pub(crate) fn will_init(raw: &[u8]) -> Result<bool, InvalidInput> {
         if raw.len() < 12 {
             return Err(InvalidInput::TooShortValue);
         }
@@ -85,7 +79,6 @@ impl ValueBytes {
 mod test {
     use super::*;
 
-    use bytes::Bytes;
     use utils::bin_ser::BeSer;
 
     macro_rules! roundtrip {
@@ -236,3 +229,56 @@ mod test {
         assert!(!ValueBytes::will_init(&expected).unwrap());
     }
 }
+
+///
+/// Result of performing GC
+///
+#[derive(Default, Serialize, Debug)]
+pub struct GcResult {
+    pub layers_total: u64,
+    pub layers_needed_by_cutoff: u64,
+    pub layers_needed_by_pitr: u64,
+    pub layers_needed_by_branches: u64,
+    pub layers_needed_by_leases: u64,
+    pub layers_not_updated: u64,
+    pub layers_removed: u64, // # of layer files removed because they have been made obsolete by newer ondisk files.
+
+    #[serde(serialize_with = "serialize_duration_as_millis")]
+    pub elapsed: Duration,
+
+    /// The layers which were garbage collected.
+    ///
+    /// Used in `/v1/tenant/:tenant_id/timeline/:timeline_id/do_gc` to wait for the layers to be
+    /// dropped in tests.
+    #[cfg(feature = "testing")]
+    #[serde(skip)]
+    pub(crate) doomed_layers: Vec<crate::tenant::storage_layer::Layer>,
+}
+
+// helper function for `GcResult`, serializing a `Duration` as an integer number of milliseconds
+fn serialize_duration_as_millis<S>(d: &Duration, serializer: S) -> Result<S::Ok, S::Error>
+where
+    S: serde::Serializer,
+{
+    d.as_millis().serialize(serializer)
+}
+
+impl AddAssign for GcResult {
+    fn add_assign(&mut self, other: Self) {
+        self.layers_total += other.layers_total;
+        self.layers_needed_by_pitr += other.layers_needed_by_pitr;
+        self.layers_needed_by_cutoff += other.layers_needed_by_cutoff;
+        self.layers_needed_by_branches += other.layers_needed_by_branches;
+        self.layers_needed_by_leases += other.layers_needed_by_leases;
+        self.layers_not_updated += other.layers_not_updated;
+        self.layers_removed += other.layers_removed;
+
+        self.elapsed += other.elapsed;
+
+        #[cfg(feature = "testing")]
+        {
+            let mut other = other;
+            self.doomed_layers.append(&mut other.doomed_layers);
+        }
+    }
+}

pageserver/src/tenant.rs

@@ -92,11 +92,11 @@ use crate::metrics::{
     remove_tenant_metrics, BROKEN_TENANTS_SET, CIRCUIT_BREAKERS_BROKEN, CIRCUIT_BREAKERS_UNBROKEN,
     TENANT_STATE_METRIC, TENANT_SYNTHETIC_SIZE_METRIC,
 };
+use crate::repository::GcResult;
 use crate::task_mgr;
 use crate::task_mgr::TaskKind;
 use crate::tenant::config::LocationMode;
 use crate::tenant::config::TenantConfOpt;
-use crate::tenant::gc_result::GcResult;
 pub use crate::tenant::remote_timeline_client::index::IndexPart;
 use crate::tenant::remote_timeline_client::remote_initdb_archive_path;
 use crate::tenant::remote_timeline_client::MaybeDeletedIndexPart;
@@ -160,7 +160,6 @@ pub(crate) mod timeline;
 pub mod size;
 
 mod gc_block;
-mod gc_result;
 pub(crate) mod throttle;
 
 pub(crate) use crate::span::debug_assert_current_span_has_tenant_and_timeline_id;
@@ -468,10 +467,10 @@ impl WalRedoManager {
     /// This method is cancellation-safe.
     pub async fn request_redo(
         &self,
-        key: pageserver_api::key::Key,
+        key: crate::repository::Key,
         lsn: Lsn,
         base_img: Option<(Lsn, bytes::Bytes)>,
-        records: Vec<(Lsn, pageserver_api::record::NeonWalRecord)>,
+        records: Vec<(Lsn, crate::walrecord::NeonWalRecord)>,
         pg_version: u32,
     ) -> Result<bytes::Bytes, walredo::Error> {
         match self {
@@ -626,10 +625,19 @@ impl TimelineOrOffloaded {
             TimelineOrOffloaded::Offloaded(offloaded) => &offloaded.delete_progress,
         }
     }
-    fn maybe_remote_client(&self) -> Option<Arc<RemoteTimelineClient>> {
+    fn remote_client_maybe_construct(&self, tenant: &Tenant) -> Arc<RemoteTimelineClient> {
         match self {
-            TimelineOrOffloaded::Timeline(timeline) => Some(timeline.remote_client.clone()),
-            TimelineOrOffloaded::Offloaded(offloaded) => offloaded.remote_client.clone(),
+            TimelineOrOffloaded::Timeline(timeline) => timeline.remote_client.clone(),
+            TimelineOrOffloaded::Offloaded(offloaded) => match offloaded.remote_client.clone() {
+                Some(remote_client) => remote_client,
+                None => {
+                    let remote_client = tenant.build_timeline_client(
+                        offloaded.timeline_id,
+                        tenant.remote_storage.clone(),
+                    );
+                    Arc::new(remote_client)
+                }
+            },
         }
     }
 }
@@ -1822,18 +1830,6 @@ impl Tenant {
         ctx: RequestContext,
     ) -> Result<Arc<Timeline>, TimelineArchivalError> {
         info!("unoffloading timeline");
-
-        // We activate the timeline below manually, so this must be called on an active timeline.
-        // We expect callers of this function to ensure this.
-        match self.current_state() {
-            TenantState::Activating { .. }
-            | TenantState::Attaching
-            | TenantState::Broken { .. } => {
-                panic!("Timeline expected to be active")
-            }
-            TenantState::Stopping { .. } => return Err(TimelineArchivalError::Cancelled),
-            TenantState::Active => {}
-        }
         let cancel = self.cancel.clone();
 
         // Protect against concurrent attempts to use this TimelineId
@@ -4810,8 +4806,7 @@ pub(crate) mod harness {
     use crate::deletion_queue::mock::MockDeletionQueue;
     use crate::l0_flush::L0FlushConfig;
     use crate::walredo::apply_neon;
-    use pageserver_api::key::Key;
-    use pageserver_api::record::NeonWalRecord;
+    use crate::{repository::Key, walrecord::NeonWalRecord};
 
     use super::*;
     use hex_literal::hex;
@@ -4858,6 +4853,7 @@ pub(crate) mod harness {
                 image_layer_creation_check_threshold: Some(
                     tenant_conf.image_layer_creation_check_threshold,
                 ),
+                switch_aux_file_policy: Some(tenant_conf.switch_aux_file_policy),
                 lsn_lease_length: Some(tenant_conf.lsn_lease_length),
                 lsn_lease_length_for_ts: Some(tenant_conf.lsn_lease_length_for_ts),
             }
@@ -5080,29 +5076,24 @@ mod tests {
 
     use super::*;
     use crate::keyspace::KeySpaceAccum;
+    use crate::repository::{Key, Value};
     use crate::tenant::harness::*;
     use crate::tenant::timeline::CompactFlags;
+    use crate::walrecord::NeonWalRecord;
     use crate::DEFAULT_PG_VERSION;
     use bytes::{Bytes, BytesMut};
     use hex_literal::hex;
     use itertools::Itertools;
-    use pageserver_api::key::{Key, AUX_KEY_PREFIX, NON_INHERITED_RANGE};
+    use pageserver_api::key::{AUX_KEY_PREFIX, NON_INHERITED_RANGE};
     use pageserver_api::keyspace::KeySpace;
     use pageserver_api::models::{CompactionAlgorithm, CompactionAlgorithmSettings};
-    use pageserver_api::value::Value;
     use rand::{thread_rng, Rng};
     use storage_layer::PersistentLayerKey;
     use tests::storage_layer::ValuesReconstructState;
     use tests::timeline::{GetVectoredError, ShutdownMode};
-    use timeline::DeltaLayerTestDesc;
-    use utils::id::TenantId;
-
-    #[cfg(feature = "testing")]
-    use pageserver_api::record::NeonWalRecord;
-    #[cfg(feature = "testing")]
     use timeline::compaction::{KeyHistoryRetention, KeyLogAtLsn};
-    #[cfg(feature = "testing")]
-    use timeline::GcInfo;
+    use timeline::{DeltaLayerTestDesc, GcInfo};
+    use utils::id::TenantId;
 
     static TEST_KEY: Lazy<Key> =
         Lazy::new(|| Key::from_slice(&hex!("010000000033333333444444445500000001")));
@@ -7668,7 +7659,6 @@ mod tests {
         Ok(())
     }
 
-    #[cfg(feature = "testing")]
     #[tokio::test]
     async fn test_neon_test_record() -> anyhow::Result<()> {
         let harness = TenantHarness::create("test_neon_test_record").await?;
@@ -7860,7 +7850,6 @@ mod tests {
         Ok(())
     }
 
-    #[cfg(feature = "testing")]
     #[tokio::test]
     async fn test_simple_bottom_most_compaction_deltas() -> anyhow::Result<()> {
         let harness = TenantHarness::create("test_simple_bottom_most_compaction_deltas").await?;
@@ -8057,7 +8046,6 @@ mod tests {
         Ok(())
     }
 
-    #[cfg(feature = "testing")]
     #[tokio::test]
     async fn test_generate_key_retention() -> anyhow::Result<()> {
         let harness = TenantHarness::create("test_generate_key_retention").await?;
@@ -8405,7 +8393,6 @@ mod tests {
         Ok(())
     }
 
-    #[cfg(feature = "testing")]
     #[tokio::test]
     async fn test_simple_bottom_most_compaction_with_retain_lsns() -> anyhow::Result<()> {
         let harness =
@@ -8646,7 +8633,6 @@ mod tests {
         Ok(())
     }
 
-    #[cfg(feature = "testing")]
     #[tokio::test]
     async fn test_simple_bottom_most_compaction_with_retain_lsns_single_key() -> anyhow::Result<()>
     {
@@ -8855,7 +8841,6 @@ mod tests {
         Ok(())
     }
 
-    #[cfg(feature = "testing")]
     #[tokio::test]
     async fn test_simple_bottom_most_compaction_on_branch() -> anyhow::Result<()> {
         let harness = TenantHarness::create("test_simple_bottom_most_compaction_on_branch").await?;
@@ -9057,7 +9042,6 @@ mod tests {
     //
     // When querying the key range [A, B) we need to read at different LSN ranges
     // for [A, C) and [C, B). This test checks that the described edge case is handled correctly.
-    #[cfg(feature = "testing")]
     #[tokio::test]
     async fn test_vectored_read_with_nested_image_layer() -> anyhow::Result<()> {
         let harness = TenantHarness::create("test_vectored_read_with_nested_image_layer").await?;

pageserver/src/tenant/config.rs

@@ -9,6 +9,7 @@
 //! may lead to a data loss.
 //!
 pub(crate) use pageserver_api::config::TenantConfigToml as TenantConf;
+use pageserver_api::models::AuxFilePolicy;
 use pageserver_api::models::CompactionAlgorithmSettings;
 use pageserver_api::models::EvictionPolicy;
 use pageserver_api::models::{self, ThrottleConfig};
@@ -340,6 +341,10 @@ pub struct TenantConfOpt {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub image_layer_creation_check_threshold: Option<u8>,
 
+    #[serde(skip_serializing_if = "Option::is_none")]
+    #[serde(default)]
+    pub switch_aux_file_policy: Option<AuxFilePolicy>,
+
     #[serde(skip_serializing_if = "Option::is_none")]
     #[serde(with = "humantime_serde")]
     #[serde(default)]
@@ -405,6 +410,9 @@ impl TenantConfOpt {
             image_layer_creation_check_threshold: self
                 .image_layer_creation_check_threshold
                 .unwrap_or(global_conf.image_layer_creation_check_threshold),
+            switch_aux_file_policy: self
+                .switch_aux_file_policy
+                .unwrap_or(global_conf.switch_aux_file_policy),
             lsn_lease_length: self
                 .lsn_lease_length
                 .unwrap_or(global_conf.lsn_lease_length),
@@ -462,6 +470,7 @@ impl From<TenantConfOpt> for models::TenantConfig {
             lazy_slru_download: value.lazy_slru_download,
             timeline_get_throttle: value.timeline_get_throttle.map(ThrottleConfig::from),
             image_layer_creation_check_threshold: value.image_layer_creation_check_threshold,
+            switch_aux_file_policy: value.switch_aux_file_policy,
             lsn_lease_length: value.lsn_lease_length.map(humantime),
             lsn_lease_length_for_ts: value.lsn_lease_length_for_ts.map(humantime),
         }

pageserver/src/tenant/gc_result.rs

@@ -1,57 +0,0 @@
-use anyhow::Result;
-use serde::Serialize;
-use std::ops::AddAssign;
-use std::time::Duration;
-
-///
-/// Result of performing GC
-///
-#[derive(Default, Serialize, Debug)]
-pub struct GcResult {
-    pub layers_total: u64,
-    pub layers_needed_by_cutoff: u64,
-    pub layers_needed_by_pitr: u64,
-    pub layers_needed_by_branches: u64,
-    pub layers_needed_by_leases: u64,
-    pub layers_not_updated: u64,
-    pub layers_removed: u64, // # of layer files removed because they have been made obsolete by newer ondisk files.
-
-    #[serde(serialize_with = "serialize_duration_as_millis")]
-    pub elapsed: Duration,
-
-    /// The layers which were garbage collected.
-    ///
-    /// Used in `/v1/tenant/:tenant_id/timeline/:timeline_id/do_gc` to wait for the layers to be
-    /// dropped in tests.
-    #[cfg(feature = "testing")]
-    #[serde(skip)]
-    pub(crate) doomed_layers: Vec<crate::tenant::storage_layer::Layer>,
-}
-
-// helper function for `GcResult`, serializing a `Duration` as an integer number of milliseconds
-fn serialize_duration_as_millis<S>(d: &Duration, serializer: S) -> Result<S::Ok, S::Error>
-where
-    S: serde::Serializer,
-{
-    d.as_millis().serialize(serializer)
-}
-
-impl AddAssign for GcResult {
-    fn add_assign(&mut self, other: Self) {
-        self.layers_total += other.layers_total;
-        self.layers_needed_by_pitr += other.layers_needed_by_pitr;
-        self.layers_needed_by_cutoff += other.layers_needed_by_cutoff;
-        self.layers_needed_by_branches += other.layers_needed_by_branches;
-        self.layers_needed_by_leases += other.layers_needed_by_leases;
-        self.layers_not_updated += other.layers_not_updated;
-        self.layers_removed += other.layers_removed;
-
-        self.elapsed += other.elapsed;
-
-        #[cfg(feature = "testing")]
-        {
-            let mut other = other;
-            self.doomed_layers.append(&mut other.doomed_layers);
-        }
-    }
-}

pageserver/src/tenant/layer_map.rs

@@ -48,9 +48,9 @@ mod layer_coverage;
 
 use crate::context::RequestContext;
 use crate::keyspace::KeyPartitioning;
+use crate::repository::Key;
 use crate::tenant::storage_layer::InMemoryLayer;
 use anyhow::Result;
-use pageserver_api::key::Key;
 use pageserver_api::keyspace::{KeySpace, KeySpaceAccum};
 use range_set_blaze::{CheckSortedDisjoint, RangeSetBlaze};
 use std::collections::{HashMap, VecDeque};

pageserver/src/tenant/mgr.rs

@@ -2811,7 +2811,7 @@ where
 }
 
 use {
-    crate::tenant::gc_result::GcResult, pageserver_api::models::TimelineGcRequest,
+    crate::repository::GcResult, pageserver_api::models::TimelineGcRequest,
     utils::http::error::ApiError,
 };
 

pageserver/src/tenant/storage_layer.rs

@@ -11,11 +11,11 @@ mod layer_name;
 pub mod merge_iterator;
 
 use crate::context::{AccessStatsBehavior, RequestContext};
+use crate::repository::Value;
+use crate::walrecord::NeonWalRecord;
 use bytes::Bytes;
 use pageserver_api::key::Key;
 use pageserver_api::keyspace::{KeySpace, KeySpaceRandomAccum};
-use pageserver_api::record::NeonWalRecord;
-use pageserver_api::value::Value;
 use std::cmp::{Ordering, Reverse};
 use std::collections::hash_map::Entry;
 use std::collections::{BinaryHeap, HashMap};

pageserver/src/tenant/storage_layer/batch_split_writer.rs

@@ -5,8 +5,7 @@ use pageserver_api::key::{Key, KEY_SIZE};
 use utils::{id::TimelineId, lsn::Lsn, shard::TenantShardId};
 
 use crate::tenant::storage_layer::Layer;
-use crate::{config::PageServerConf, context::RequestContext, tenant::Timeline};
-use pageserver_api::value::Value;
+use crate::{config::PageServerConf, context::RequestContext, repository::Value, tenant::Timeline};
 
 use super::layer::S3_UPLOAD_LIMIT;
 use super::{

pageserver/src/tenant/storage_layer/delta_layer.rs

@@ -30,6 +30,7 @@
 use crate::config::PageServerConf;
 use crate::context::{PageContentKind, RequestContext, RequestContextBuilder};
 use crate::page_cache::{self, FileId, PAGE_SZ};
+use crate::repository::{Key, Value, KEY_SIZE};
 use crate::tenant::blob_io::BlobWriter;
 use crate::tenant::block_io::{BlockBuf, BlockCursor, BlockLease, BlockReader, FileBlockReader};
 use crate::tenant::disk_btree::{
@@ -45,7 +46,7 @@ use crate::tenant::PageReconstructError;
 use crate::virtual_file::owned_buffers_io::io_buf_ext::{FullSlice, IoBufExt};
 use crate::virtual_file::IoBufferMut;
 use crate::virtual_file::{self, MaybeFatalIo, VirtualFile};
-use crate::TEMP_FILE_SUFFIX;
+use crate::{walrecord, TEMP_FILE_SUFFIX};
 use crate::{DELTA_FILE_MAGIC, STORAGE_FORMAT_VERSION};
 use anyhow::{anyhow, bail, ensure, Context, Result};
 use camino::{Utf8Path, Utf8PathBuf};
@@ -53,11 +54,9 @@ use futures::StreamExt;
 use itertools::Itertools;
 use pageserver_api::config::MaxVectoredReadBytes;
 use pageserver_api::key::DBDIR_KEY;
-use pageserver_api::key::{Key, KEY_SIZE};
 use pageserver_api::keyspace::KeySpace;
 use pageserver_api::models::ImageCompressionAlgorithm;
 use pageserver_api::shard::TenantShardId;
-use pageserver_api::value::Value;
 use rand::{distributions::Alphanumeric, Rng};
 use serde::{Deserialize, Serialize};
 use std::collections::VecDeque;
@@ -1294,7 +1293,7 @@ impl DeltaLayerInner {
                     // is it an image or will_init walrecord?
                     // FIXME: this could be handled by threading the BlobRef to the
                     // VectoredReadBuilder
-                    let will_init = pageserver_api::value::ValueBytes::will_init(&data)
+                    let will_init = crate::repository::ValueBytes::will_init(&data)
                         .inspect_err(|_e| {
                             #[cfg(feature = "testing")]
                             tracing::error!(data=?utils::Hex(&data), err=?_e, %key, %lsn, "failed to parse will_init out of serialized value");
@@ -1357,7 +1356,7 @@ impl DeltaLayerInner {
                     format!(" img {} bytes", img.len())
                 }
                 Value::WalRecord(rec) => {
-                    let wal_desc = pageserver_api::record::describe_wal_record(&rec)?;
+                    let wal_desc = walrecord::describe_wal_record(&rec)?;
                     format!(
                         " rec {} bytes will_init: {} {}",
                         buf.len(),
@@ -1611,6 +1610,7 @@ pub(crate) mod test {
     use rand::RngCore;
 
     use super::*;
+    use crate::repository::Value;
     use crate::tenant::harness::TIMELINE_ID;
     use crate::tenant::storage_layer::{Layer, ResidentLayer};
     use crate::tenant::vectored_blob_io::StreamingVectoredReadPlanner;
@@ -1622,7 +1622,6 @@ pub(crate) mod test {
         DEFAULT_PG_VERSION,
     };
     use bytes::Bytes;
-    use pageserver_api::value::Value;
 
     /// Construct an index for a fictional delta layer and and then
     /// traverse in order to plan vectored reads for a query. Finally,
@@ -1975,8 +1974,8 @@ pub(crate) mod test {
 
     #[tokio::test]
     async fn copy_delta_prefix_smoke() {
+        use crate::walrecord::NeonWalRecord;
         use bytes::Bytes;
-        use pageserver_api::record::NeonWalRecord;
 
         let h = crate::tenant::harness::TenantHarness::create("truncate_delta_smoke")
             .await
@@ -2199,7 +2198,6 @@ pub(crate) mod test {
         (k1, l1).cmp(&(k2, l2))
     }
 
-    #[cfg(feature = "testing")]
     pub(crate) fn sort_delta_value(
         (k1, l1, v1): &(Key, Lsn, Value),
         (k2, l2, v2): &(Key, Lsn, Value),

pageserver/src/tenant/storage_layer/filter_iterator.rs

@@ -7,7 +7,7 @@ use pageserver_api::{
 };
 use utils::lsn::Lsn;
 
-use pageserver_api::value::Value;
+use crate::repository::Value;
 
 use super::merge_iterator::MergeIterator;
 
@@ -121,8 +121,8 @@ mod tests {
 
     #[tokio::test]
     async fn filter_keyspace_iterator() {
+        use crate::repository::Value;
         use bytes::Bytes;
-        use pageserver_api::value::Value;
 
         let harness = TenantHarness::create("filter_iterator_filter_keyspace_iterator")
             .await

pageserver/src/tenant/storage_layer/image_layer.rs

@@ -28,6 +28,7 @@
 use crate::config::PageServerConf;
 use crate::context::{PageContentKind, RequestContext, RequestContextBuilder};
 use crate::page_cache::{self, FileId, PAGE_SZ};
+use crate::repository::{Key, Value, KEY_SIZE};
 use crate::tenant::blob_io::BlobWriter;
 use crate::tenant::block_io::{BlockBuf, FileBlockReader};
 use crate::tenant::disk_btree::{
@@ -50,10 +51,8 @@ use hex;
 use itertools::Itertools;
 use pageserver_api::config::MaxVectoredReadBytes;
 use pageserver_api::key::DBDIR_KEY;
-use pageserver_api::key::{Key, KEY_SIZE};
 use pageserver_api::keyspace::KeySpace;
 use pageserver_api::shard::{ShardIdentity, TenantShardId};
-use pageserver_api::value::Value;
 use rand::{distributions::Alphanumeric, Rng};
 use serde::{Deserialize, Serialize};
 use std::collections::VecDeque;
@@ -1126,7 +1125,6 @@ mod test {
     use pageserver_api::{
         key::Key,
         shard::{ShardCount, ShardIdentity, ShardNumber, ShardStripeSize},
-        value::Value,
     };
     use utils::{
         generation::Generation,
@@ -1136,6 +1134,7 @@ mod test {
 
     use crate::{
         context::RequestContext,
+        repository::Value,
         tenant::{
             config::TenantConf,
             harness::{TenantHarness, TIMELINE_ID},

pageserver/src/tenant/storage_layer/inmemory_layer.rs

@@ -7,6 +7,7 @@
 use crate::assert_u64_eq_usize::{u64_to_usize, U64IsUsize, UsizeIsU64};
 use crate::config::PageServerConf;
 use crate::context::{PageContentKind, RequestContext, RequestContextBuilder};
+use crate::repository::{Key, Value};
 use crate::tenant::ephemeral_file::EphemeralFile;
 use crate::tenant::timeline::GetVectoredError;
 use crate::tenant::PageReconstructError;
@@ -15,11 +16,9 @@ use crate::{l0_flush, page_cache};
 use anyhow::{anyhow, Context, Result};
 use camino::Utf8PathBuf;
 use pageserver_api::key::CompactKey;
-use pageserver_api::key::Key;
 use pageserver_api::keyspace::KeySpace;
 use pageserver_api::models::InMemoryLayerInfo;
 use pageserver_api::shard::TenantShardId;
-use pageserver_api::value::Value;
 use std::collections::{BTreeMap, HashMap};
 use std::sync::{Arc, OnceLock};
 use std::time::Instant;

pageserver/src/tenant/storage_layer/layer/tests.rs

@@ -760,8 +760,8 @@ async fn evict_and_wait_does_not_wait_for_download() {
 /// Also checks that the same does not happen on a non-evicted layer (regression test).
 #[tokio::test(start_paused = true)]
 async fn eviction_cancellation_on_drop() {
+    use crate::repository::Value;
     use bytes::Bytes;
-    use pageserver_api::value::Value;
 
     // this is the runtime on which Layer spawns the blocking tasks on
     let handle = tokio::runtime::Handle::current();
@@ -782,7 +782,7 @@ async fn eviction_cancellation_on_drop() {
         let mut writer = timeline.writer().await;
         writer
             .put(
-                pageserver_api::key::Key::from_i128(5),
+                crate::repository::Key::from_i128(5),
                 Lsn(0x20),
                 &Value::Image(Bytes::from_static(b"this does not matter either")),
                 &ctx,

pageserver/src/tenant/storage_layer/layer_desc.rs

@@ -3,7 +3,7 @@ use pageserver_api::shard::TenantShardId;
 use std::ops::Range;
 use utils::{id::TimelineId, lsn::Lsn};
 
-use pageserver_api::key::Key;
+use crate::repository::Key;
 
 use super::{DeltaLayerName, ImageLayerName, LayerName};
 

pageserver/src/tenant/storage_layer/layer_name.rs

@@ -1,7 +1,7 @@
 //!
 //! Helper functions for dealing with filenames of the image and delta layer files.
 //!
-use pageserver_api::key::Key;
+use crate::repository::Key;
 use std::borrow::Cow;
 use std::cmp::Ordering;
 use std::fmt;

pageserver/src/tenant/storage_layer/merge_iterator.rs

@@ -7,8 +7,7 @@ use anyhow::bail;
 use pageserver_api::key::Key;
 use utils::lsn::Lsn;
 
-use crate::context::RequestContext;
-use pageserver_api::value::Value;
+use crate::{context::RequestContext, repository::Value};
 
 use super::{
     delta_layer::{DeltaLayerInner, DeltaLayerIterator},
@@ -292,16 +291,12 @@ mod tests {
     use crate::{
         tenant::{
             harness::{TenantHarness, TIMELINE_ID},
-            storage_layer::delta_layer::test::{produce_delta_layer, sort_delta},
+            storage_layer::delta_layer::test::{produce_delta_layer, sort_delta, sort_delta_value},
         },
+        walrecord::NeonWalRecord,
         DEFAULT_PG_VERSION,
     };
 
-    #[cfg(feature = "testing")]
-    use crate::tenant::storage_layer::delta_layer::test::sort_delta_value;
-    #[cfg(feature = "testing")]
-    use pageserver_api::record::NeonWalRecord;
-
     async fn assert_merge_iter_equal(
         merge_iter: &mut MergeIterator<'_>,
         expect: &[(Key, Lsn, Value)],
@@ -324,8 +319,8 @@ mod tests {
 
     #[tokio::test]
     async fn merge_in_between() {
+        use crate::repository::Value;
         use bytes::Bytes;
-        use pageserver_api::value::Value;
 
         let harness = TenantHarness::create("merge_iterator_merge_in_between")
             .await
@@ -389,8 +384,8 @@ mod tests {
 
     #[tokio::test]
     async fn delta_merge() {
+        use crate::repository::Value;
         use bytes::Bytes;
-        use pageserver_api::value::Value;
 
         let harness = TenantHarness::create("merge_iterator_delta_merge")
             .await
@@ -463,11 +458,10 @@ mod tests {
         // TODO: test layers are loaded only when needed, reducing num of active iterators in k-merge
     }
 
-    #[cfg(feature = "testing")]
     #[tokio::test]
     async fn delta_image_mixed_merge() {
+        use crate::repository::Value;
         use bytes::Bytes;
-        use pageserver_api::value::Value;
 
         let harness = TenantHarness::create("merge_iterator_delta_image_mixed_merge")
             .await
@@ -592,6 +586,5 @@ mod tests {
         is_send(merge_iter);
     }
 
-    #[cfg(feature = "testing")]
     fn is_send(_: impl Send) {}
 }

pageserver/src/tenant/timeline.rs

@@ -125,12 +125,11 @@ use utils::{
     simple_rcu::{Rcu, RcuReadGuard},
 };
 
+use crate::repository::GcResult;
+use crate::repository::{Key, Value};
 use crate::task_mgr;
 use crate::task_mgr::TaskKind;
-use crate::tenant::gc_result::GcResult;
 use crate::ZERO_PAGE;
-use pageserver_api::key::Key;
-use pageserver_api::value::Value;
 
 use self::delete::DeleteTimelineFlow;
 pub(super) use self::eviction_task::EvictionTaskTenantState;
@@ -5823,15 +5822,17 @@ fn is_send() {
 #[cfg(test)]
 mod tests {
     use pageserver_api::key::Key;
-    use pageserver_api::value::Value;
     use utils::{id::TimelineId, lsn::Lsn};
 
-    use crate::tenant::{
-        harness::{test_img, TenantHarness},
-        layer_map::LayerMap,
-        storage_layer::{Layer, LayerName},
-        timeline::{DeltaLayerTestDesc, EvictionError},
-        Timeline,
+    use crate::{
+        repository::Value,
+        tenant::{
+            harness::{test_img, TenantHarness},
+            layer_map::LayerMap,
+            storage_layer::{Layer, LayerName},
+            timeline::{DeltaLayerTestDesc, EvictionError},
+            Timeline,
+        },
     };
 
     #[tokio::test]

pageserver/src/tenant/timeline/compaction.rs

@@ -49,10 +49,9 @@ use pageserver_api::config::tenant_conf_defaults::{
     DEFAULT_CHECKPOINT_DISTANCE, DEFAULT_COMPACTION_THRESHOLD,
 };
 
-use pageserver_api::key::Key;
-use pageserver_api::keyspace::KeySpace;
-use pageserver_api::record::NeonWalRecord;
-use pageserver_api::value::Value;
+use crate::keyspace::KeySpace;
+use crate::repository::{Key, Value};
+use crate::walrecord::NeonWalRecord;
 
 use utils::lsn::Lsn;
 
@@ -2149,7 +2148,7 @@ struct ResidentDeltaLayer(ResidentLayer);
 struct ResidentImageLayer(ResidentLayer);
 
 impl CompactionJobExecutor for TimelineAdaptor {
-    type Key = pageserver_api::key::Key;
+    type Key = crate::repository::Key;
 
     type Layer = OwnArc<PersistentLayerDesc>;
     type DeltaLayer = ResidentDeltaLayer;

pageserver/src/tenant/timeline/delete.rs

@@ -6,7 +6,7 @@ use std::{
 use anyhow::Context;
 use pageserver_api::{models::TimelineState, shard::TenantShardId};
 use tokio::sync::OwnedMutexGuard;
-use tracing::{error, info, info_span, instrument, Instrument};
+use tracing::{error, info, instrument, Instrument};
 use utils::{crashsafe, fs_ext, id::TimelineId, pausable_failpoint};
 
 use crate::{
@@ -15,7 +15,7 @@ use crate::{
     tenant::{
         metadata::TimelineMetadata,
         remote_timeline_client::{
-            self, MaybeDeletedIndexPart, PersistIndexPartWithDeletedFlagError, RemoteTimelineClient,
+            self, PersistIndexPartWithDeletedFlagError, RemoteTimelineClient,
         },
         CreateTimelineCause, DeleteTimelineError, Tenant, TimelineOrOffloaded,
     },
@@ -258,32 +258,7 @@ impl DeleteTimelineFlow {
             ))?
         });
 
-        let remote_client = match timeline.maybe_remote_client() {
-            Some(remote_client) => remote_client,
-            None => {
-                let remote_client = tenant
-                    .build_timeline_client(timeline.timeline_id(), tenant.remote_storage.clone());
-                let result = remote_client
-                    .download_index_file(&tenant.cancel)
-                    .instrument(info_span!("download_index_file"))
-                    .await
-                    .map_err(|e| DeleteTimelineError::Other(anyhow::anyhow!("error: {:?}", e)))?;
-                let index_part = match result {
-                    MaybeDeletedIndexPart::Deleted(p) => {
-                        tracing::info!("Timeline already set as deleted in remote index");
-                        p
-                    }
-                    MaybeDeletedIndexPart::IndexPart(p) => p,
-                };
-                let remote_client = Arc::new(remote_client);
-
-                remote_client
-                    .init_upload_queue(&index_part)
-                    .map_err(DeleteTimelineError::Other)?;
-                remote_client.shutdown().await;
-                remote_client
-            }
-        };
+        let remote_client = timeline.remote_client_maybe_construct(tenant);
         set_deleted_in_remote_index(&remote_client).await?;
 
         fail::fail_point!("timeline-delete-before-schedule", |_| {

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

@@ -31,11 +31,11 @@ use crate::{
     task_mgr::{TaskKind, WALRECEIVER_RUNTIME},
     tenant::{debug_assert_current_span_has_tenant_and_timeline_id, Timeline, WalReceiverInfo},
     walingest::WalIngest,
+    walrecord::{decode_wal_record, DecodedWALRecord},
 };
 use postgres_backend::is_expected_io_error;
 use postgres_connection::PgConnectionConfig;
 use postgres_ffi::waldecoder::WalStreamDecoder;
-use postgres_ffi::walrecord::{decode_wal_record, DecodedWALRecord};
 use utils::{id::NodeId, lsn::Lsn};
 use utils::{pageserver_feedback::PageserverFeedback, sync::gate::GateError};
 

pageserver/src/walingest.rs

@@ -29,10 +29,8 @@ use std::time::Instant;
 use std::time::SystemTime;
 
 use pageserver_api::shard::ShardIdentity;
-use postgres_ffi::walrecord::*;
 use postgres_ffi::{dispatch_pgversion, enum_pgversion, enum_pgversion_dispatch, TimestampTz};
 use postgres_ffi::{fsm_logical_to_physical, page_is_new, page_set_lsn};
-use wal_decoder::models::*;
 
 use anyhow::{bail, Context, Result};
 use bytes::{Buf, Bytes, BytesMut};
@@ -46,9 +44,9 @@ use crate::pgdatadir_mapping::{DatadirModification, Version};
 use crate::span::debug_assert_current_span_has_tenant_and_timeline_id;
 use crate::tenant::PageReconstructError;
 use crate::tenant::Timeline;
+use crate::walrecord::*;
 use crate::ZERO_PAGE;
 use pageserver_api::key::rel_block_to_key;
-use pageserver_api::record::NeonWalRecord;
 use pageserver_api::reltag::{BlockNumber, RelTag, SlruKind};
 use postgres_ffi::pg_constants;
 use postgres_ffi::relfile_utils::{FSM_FORKNUM, INIT_FORKNUM, MAIN_FORKNUM, VISIBILITYMAP_FORKNUM};
@@ -110,6 +108,143 @@ struct WarnIngestLag {
     timestamp_invalid_msg_ratelimit: RateLimit,
 }
 
+// These structs are an intermediary representation of the PostgreSQL WAL records.
+// The ones prefixed with `Xl` are lower level, while the ones that are not have
+// all the required context to be acted upon by the pageserver.
+
+enum HeapamRecord {
+    ClearVmBits(ClearVmBits),
+}
+
+struct ClearVmBits {
+    new_heap_blkno: Option<u32>,
+    old_heap_blkno: Option<u32>,
+    vm_rel: RelTag,
+    flags: u8,
+}
+
+enum NeonrmgrRecord {
+    ClearVmBits(ClearVmBits),
+}
+
+enum SmgrRecord {
+    Create(SmgrCreate),
+    Truncate(XlSmgrTruncate),
+}
+
+struct SmgrCreate {
+    rel: RelTag,
+}
+
+enum DbaseRecord {
+    Create(DbaseCreate),
+    Drop(DbaseDrop),
+}
+
+struct DbaseCreate {
+    db_id: u32,
+    tablespace_id: u32,
+    src_db_id: u32,
+    src_tablespace_id: u32,
+}
+
+struct DbaseDrop {
+    db_id: u32,
+    tablespace_ids: Vec<u32>,
+}
+
+enum ClogRecord {
+    ZeroPage(ClogZeroPage),
+    Truncate(ClogTruncate),
+}
+
+struct ClogZeroPage {
+    segno: u32,
+    rpageno: u32,
+}
+
+struct ClogTruncate {
+    pageno: u32,
+    oldest_xid: u32,
+    oldest_xid_db: u32,
+}
+
+enum XactRecord {
+    Commit(XactCommon),
+    Abort(XactCommon),
+    CommitPrepared(XactCommon),
+    AbortPrepared(XactCommon),
+    Prepare(XactPrepare),
+}
+
+struct XactCommon {
+    parsed: XlXactParsedRecord,
+    origin_id: u16,
+    // Fields below are only used for logging
+    xl_xid: u32,
+    lsn: Lsn,
+}
+
+struct XactPrepare {
+    xl_xid: u32,
+    data: Bytes,
+}
+
+enum MultiXactRecord {
+    ZeroPage(MultiXactZeroPage),
+    Create(XlMultiXactCreate),
+    Truncate(XlMultiXactTruncate),
+}
+
+struct MultiXactZeroPage {
+    slru_kind: SlruKind,
+    segno: u32,
+    rpageno: u32,
+}
+
+enum RelmapRecord {
+    Update(RelmapUpdate),
+}
+
+struct RelmapUpdate {
+    update: XlRelmapUpdate,
+    buf: Bytes,
+}
+
+enum XlogRecord {
+    Raw(RawXlogRecord),
+}
+
+struct RawXlogRecord {
+    info: u8,
+    lsn: Lsn,
+    buf: Bytes,
+}
+
+enum LogicalMessageRecord {
+    Put(PutLogicalMessage),
+    #[cfg(feature = "testing")]
+    Failpoint,
+}
+
+struct PutLogicalMessage {
+    path: String,
+    buf: Bytes,
+}
+
+enum StandbyRecord {
+    RunningXacts(StandbyRunningXacts),
+}
+
+struct StandbyRunningXacts {
+    oldest_running_xid: u32,
+}
+
+enum ReploriginRecord {
+    Set(XlReploriginSet),
+    Drop(XlReploriginDrop),
+}
+
 impl WalIngest {
     pub async fn new(
         timeline: &Timeline,
@@ -149,6 +284,7 @@ impl WalIngest {
     /// relations/pages that the record affects.
     ///
     /// This function returns `true` if the record was ingested, and `false` if it was filtered out
+    ///
     pub async fn ingest_record(
         &mut self,
         decoded: DecodedWALRecord,
@@ -2082,7 +2218,7 @@ impl WalIngest {
     ) -> anyhow::Result<Option<LogicalMessageRecord>> {
         let info = decoded.xl_info & pg_constants::XLR_RMGR_INFO_MASK;
         if info == pg_constants::XLOG_LOGICAL_MESSAGE {
-            let xlrec = XlLogicalMessage::decode(buf);
+            let xlrec = crate::walrecord::XlLogicalMessage::decode(buf);
             let prefix = std::str::from_utf8(&buf[0..xlrec.prefix_size - 1])?;
 
             #[cfg(feature = "testing")]
@@ -2110,7 +2246,7 @@ impl WalIngest {
     ) -> anyhow::Result<Option<StandbyRecord>> {
         let info = decoded.xl_info & pg_constants::XLR_RMGR_INFO_MASK;
         if info == pg_constants::XLOG_RUNNING_XACTS {
-            let xlrec = XlRunningXacts::decode(buf);
+            let xlrec = crate::walrecord::XlRunningXacts::decode(buf);
             return Ok(Some(StandbyRecord::RunningXacts(StandbyRunningXacts {
                 oldest_running_xid: xlrec.oldest_running_xid,
             })));
@@ -2140,10 +2276,10 @@ impl WalIngest {
     ) -> anyhow::Result<Option<ReploriginRecord>> {
         let info = decoded.xl_info & pg_constants::XLR_RMGR_INFO_MASK;
         if info == pg_constants::XLOG_REPLORIGIN_SET {
-            let xlrec = XlReploriginSet::decode(buf);
+            let xlrec = crate::walrecord::XlReploriginSet::decode(buf);
             return Ok(Some(ReploriginRecord::Set(xlrec)));
         } else if info == pg_constants::XLOG_REPLORIGIN_DROP {
-            let xlrec = XlReploriginDrop::decode(buf);
+            let xlrec = crate::walrecord::XlReploriginDrop::decode(buf);
             return Ok(Some(ReploriginRecord::Drop(xlrec)));
         }
 
@@ -3010,7 +3146,6 @@ mod tests {
     async fn test_ingest_real_wal() {
         use crate::tenant::harness::*;
         use postgres_ffi::waldecoder::WalStreamDecoder;
-        use postgres_ffi::walrecord::decode_wal_record;
         use postgres_ffi::WAL_SEGMENT_SIZE;
 
         // Define test data path and constants.

pageserver/src/walredo.rs

@@ -29,11 +29,11 @@ use crate::metrics::{
     WAL_REDO_BYTES_HISTOGRAM, WAL_REDO_PROCESS_LAUNCH_DURATION_HISTOGRAM,
     WAL_REDO_RECORDS_HISTOGRAM, WAL_REDO_TIME,
 };
+use crate::repository::Key;
+use crate::walrecord::NeonWalRecord;
 use anyhow::Context;
 use bytes::{Bytes, BytesMut};
-use pageserver_api::key::Key;
 use pageserver_api::models::{WalRedoManagerProcessStatus, WalRedoManagerStatus};
-use pageserver_api::record::NeonWalRecord;
 use pageserver_api::shard::TenantShardId;
 use std::future::Future;
 use std::sync::Arc;
@@ -548,10 +548,9 @@ impl PostgresRedoManager {
 #[cfg(test)]
 mod tests {
     use super::PostgresRedoManager;
-    use crate::config::PageServerConf;
+    use crate::repository::Key;
+    use crate::{config::PageServerConf, walrecord::NeonWalRecord};
     use bytes::Bytes;
-    use pageserver_api::key::Key;
-    use pageserver_api::record::NeonWalRecord;
     use pageserver_api::shard::TenantShardId;
     use std::str::FromStr;
     use tracing::Instrument;

pageserver/src/walredo/apply_neon.rs

@@ -1,8 +1,8 @@
+use crate::walrecord::NeonWalRecord;
 use anyhow::Context;
 use byteorder::{ByteOrder, LittleEndian};
 use bytes::BytesMut;
 use pageserver_api::key::Key;
-use pageserver_api::record::NeonWalRecord;
 use pageserver_api::reltag::SlruKind;
 use postgres_ffi::pg_constants;
 use postgres_ffi::relfile_utils::VISIBILITYMAP_FORKNUM;
@@ -238,7 +238,7 @@ pub(crate) fn apply_in_neon(
             // No-op: this record will never be created in aux v2.
             warn!("AuxFile record should not be created in aux v2");
         }
-        #[cfg(feature = "testing")]
+        #[cfg(test)]
         NeonWalRecord::Test {
             append,
             clear,

pageserver/src/walredo/process.rs

@@ -8,10 +8,10 @@ use crate::{
     metrics::{WalRedoKillCause, WAL_REDO_PROCESS_COUNTERS, WAL_REDO_RECORD_COUNTER},
     page_cache::PAGE_SZ,
     span::debug_assert_current_span_has_tenant_id,
+    walrecord::NeonWalRecord,
 };
 use anyhow::Context;
 use bytes::Bytes;
-use pageserver_api::record::NeonWalRecord;
 use pageserver_api::{reltag::RelTag, shard::TenantShardId};
 use postgres_ffi::BLCKSZ;
 #[cfg(feature = "testing")]

pgxn/neon/Makefile

@@ -16,7 +16,6 @@ OBJS = \
 	neon_walreader.o \
 	pagestore_smgr.o \
 	relsize_cache.o \
-	unstable_extensions.o \
 	walproposer.o \
 	walproposer_pg.o \
 	control_plane_connector.o \

pgxn/neon/control_plane_connector.c

@@ -18,7 +18,6 @@
  *
  *-------------------------------------------------------------------------
  */
-
 #include "postgres.h"
 
 #include <curl/curl.h>
@@ -509,8 +508,6 @@ NeonXactCallback(XactEvent event, void *arg)
 static bool
 RoleIsNeonSuperuser(const char *role_name)
 {
-	Assert(role_name);
-
 	return strcmp(role_name, "neon_superuser") == 0;
 }
 
@@ -673,7 +670,7 @@ HandleCreateRole(CreateRoleStmt *stmt)
 static void
 HandleAlterRole(AlterRoleStmt *stmt)
 {
-	char	   *role_name;
+	const char *role_name = stmt->role->rolename;
 	DefElem    *dpass;
 	ListCell   *option;
 	bool		found = false;
@@ -681,7 +678,6 @@ HandleAlterRole(AlterRoleStmt *stmt)
 
 	InitRoleTableIfNeeded();
 
-	role_name = get_rolespec_name(stmt->role);
 	if (RoleIsNeonSuperuser(role_name) && !superuser())
 		elog(ERROR, "can't ALTER neon_superuser");
 
@@ -693,13 +689,9 @@ HandleAlterRole(AlterRoleStmt *stmt)
 		if (strcmp(defel->defname, "password") == 0)
 			dpass = defel;
 	}
-
 	/* We only care about updates to the password */
 	if (!dpass)
-	{
-		pfree(role_name);
 		return;
-	}
 
 	entry = hash_search(CurrentDdlTable->role_table,
 						role_name,
@@ -712,8 +704,6 @@ HandleAlterRole(AlterRoleStmt *stmt)
 	else
 		entry->password = NULL;
 	entry->type = Op_Set;
-
-	pfree(role_name);
 }
 
 static void

pgxn/neon/neon.c

@@ -30,7 +30,6 @@
 #include "neon.h"
 #include "control_plane_connector.h"
 #include "logical_replication_monitor.h"
-#include "unstable_extensions.h"
 #include "walsender_hooks.h"
 #if PG_MAJORVERSION_NUM >= 16
 #include "storage/ipc.h"
@@ -425,7 +424,6 @@ _PG_init(void)
 	LogicalFuncs_Custom_XLogReaderRoutines = NeonOnDemandXLogReaderRoutines;
 	SlotFuncs_Custom_XLogReaderRoutines = NeonOnDemandXLogReaderRoutines;
 
-	InitUnstableExtensionsSupport();
 	InitLogicalReplicationMonitor();
 	InitControlPlaneConnector();
 

pgxn/neon/neon_pgversioncompat.c

@@ -42,4 +42,3 @@ InitMaterializedSRF(FunctionCallInfo fcinfo, bits32 flags)
 	MemoryContextSwitchTo(old_context);
 }
 #endif
-

pgxn/neon/unstable_extensions.c

@@ -1,129 +0,0 @@
-#include <stdlib.h>
-#include <string.h>
-
-#include "postgres.h"
-
-#include "nodes/plannodes.h"
-#include "nodes/parsenodes.h"
-#include "tcop/utility.h"
-#include "utils/errcodes.h"
-#include "utils/guc.h"
-
-#include "neon_pgversioncompat.h"
-#include "unstable_extensions.h"
-
-static bool					allow_unstable_extensions = false;
-static char				   *unstable_extensions = NULL;
-
-static ProcessUtility_hook_type PreviousProcessUtilityHook = NULL;
-
-static bool
-list_contains(char const* comma_separated_list, char const* val)
-{
-	char const* occ = comma_separated_list;
-	size_t val_len = strlen(val);
-
-	if (val_len == 0)
-		return false;
-
-	while ((occ = strstr(occ, val)) != NULL)
-	{
-		if ((occ == comma_separated_list || occ[-1] == ',')
-			&& (occ[val_len] == '\0' || occ[val_len] == ','))
-		{
-			return true;
-		}
-		occ += val_len;
-	}
-
-	return false;
-}
-
-
-static void
-CheckUnstableExtension(
-	PlannedStmt *pstmt,
-	const char *queryString,
-	bool readOnlyTree,
-	ProcessUtilityContext context,
-	ParamListInfo params,
-	QueryEnvironment *queryEnv,
-	DestReceiver *dest,
-	QueryCompletion *qc)
-{
-	Node	   *parseTree = pstmt->utilityStmt;
-
-	if (allow_unstable_extensions || unstable_extensions == NULL)
-		goto process;
-
-	switch (nodeTag(parseTree))
-	{
-		case T_CreateExtensionStmt:
-		{
-			CreateExtensionStmt *stmt = castNode(CreateExtensionStmt, parseTree);
-			if (list_contains(unstable_extensions, stmt->extname))
-			{
-				ereport(ERROR,
-						(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
-						 errmsg("installing %s is currently prohibited", stmt->extname),
-						 errhint("Set neon.allow_unstable_extensions to true")));
-			}
-			break;
-		}
-		default:
-			goto process;
-	}
-
-process:
-	if (PreviousProcessUtilityHook)
-	{
-		PreviousProcessUtilityHook(
-			pstmt,
-			queryString,
-			readOnlyTree,
-			context,
-			params,
-			queryEnv,
-			dest,
-			qc);
-	}
-	else
-	{
-		standard_ProcessUtility(
-			pstmt,
-			queryString,
-			readOnlyTree,
-			context,
-			params,
-			queryEnv,
-			dest,
-			qc);
-	}
-}
-
-void
-InitUnstableExtensionsSupport(void)
-{
-	DefineCustomBoolVariable(
-		"neon.allow_unstable_extensions",
-		"Allow unstable extensions to be installed and used",
-		NULL,
-		&allow_unstable_extensions,
-		false,
-		PGC_SUSET,
-		0,
-		NULL, NULL, NULL);
-
-	DefineCustomStringVariable(
-		"neon.unstable_extensions",
-		"Allow unstable extensions to be installed and used",
-		NULL,
-		&unstable_extensions,
-		NULL,
-		PGC_SUSET,
-		0,
-		NULL, NULL, NULL);
-
-	PreviousProcessUtilityHook = ProcessUtility_hook;
-	ProcessUtility_hook = CheckUnstableExtension;
-}

pgxn/neon/unstable_extensions.h

@@ -1,6 +0,0 @@
-#ifndef __NEON_UNSTABLE_EXTENSIONS_H__
-#define __NEON_UNSTABLE_EXTENSIONS_H__
-
-void InitUnstableExtensionsSupport(void);
-
-#endif

proxy/src/auth/backend/jwt.rs

@@ -1,4 +1,3 @@
-use std::borrow::Cow;
 use std::future::Future;
 use std::sync::Arc;
 use std::time::{Duration, SystemTime};
@@ -46,7 +45,6 @@ pub(crate) enum FetchAuthRulesError {
     RoleJwksNotConfigured,
 }
 
-#[derive(Clone)]
 pub(crate) struct AuthRule {
     pub(crate) id: String,
     pub(crate) jwks_url: url::Url,
@@ -279,7 +277,7 @@ impl JwkCacheEntryLock {
 
         // get the key from the JWKs if possible. If not, wait for the keys to update.
         let (jwk, expected_audience) = loop {
-            match guard.find_jwk_and_audience(&kid, role_name) {
+            match guard.find_jwk_and_audience(kid, role_name) {
                 Some(jwk) => break jwk,
                 None if guard.last_retrieved.elapsed() > MIN_RENEW => {
                     let _paused = ctx.latency_timer_pause(crate::metrics::Waiting::Compute);
@@ -314,9 +312,7 @@ impl JwkCacheEntryLock {
 
         if let Some(aud) = expected_audience {
             if payload.audience.0.iter().all(|s| s != aud) {
-                return Err(JwtError::InvalidClaims(
-                    JwtClaimsError::InvalidJwtTokenAudience,
-                ));
+                return Err(JwtError::InvalidJwtTokenAudience);
             }
         }
 
@@ -324,15 +320,13 @@ impl JwkCacheEntryLock {
 
         if let Some(exp) = payload.expiration {
             if now >= exp + CLOCK_SKEW_LEEWAY {
-                return Err(JwtError::InvalidClaims(JwtClaimsError::JwtTokenHasExpired));
+                return Err(JwtError::JwtTokenHasExpired);
             }
         }
 
         if let Some(nbf) = payload.not_before {
             if nbf >= now + CLOCK_SKEW_LEEWAY {
-                return Err(JwtError::InvalidClaims(
-                    JwtClaimsError::JwtTokenNotYetReadyToUse,
-                ));
+                return Err(JwtError::JwtTokenNotYetReadyToUse);
             }
         }
 
@@ -426,8 +420,8 @@ struct JwtHeader<'a> {
     #[serde(rename = "alg")]
     algorithm: jose_jwa::Algorithm,
     /// key id, must be provided for our usecase
-    #[serde(rename = "kid", borrow)]
-    key_id: Option<Cow<'a, str>>,
+    #[serde(rename = "kid")]
+    key_id: Option<&'a str>,
 }
 
 /// <https://datatracker.ietf.org/doc/html/rfc7519#section-4.1>
@@ -446,17 +440,17 @@ struct JwtPayload<'a> {
 
     // the following entries are only extracted for the sake of debug logging.
     /// Issuer of the JWT
-    #[serde(rename = "iss", borrow)]
-    issuer: Option<Cow<'a, str>>,
+    #[serde(rename = "iss")]
+    issuer: Option<&'a str>,
     /// Subject of the JWT (the user)
-    #[serde(rename = "sub", borrow)]
-    subject: Option<Cow<'a, str>>,
+    #[serde(rename = "sub")]
+    subject: Option<&'a str>,
     /// Unique token identifier
-    #[serde(rename = "jti", borrow)]
-    jwt_id: Option<Cow<'a, str>>,
+    #[serde(rename = "jti")]
+    jwt_id: Option<&'a str>,
     /// Unique session identifier
-    #[serde(rename = "sid", borrow)]
-    session_id: Option<Cow<'a, str>>,
+    #[serde(rename = "sid")]
+    session_id: Option<&'a str>,
 }
 
 /// `OneOrMany` supports parsing either a single item or an array of items.
@@ -591,8 +585,14 @@ pub(crate) enum JwtError {
     #[error("Provided authentication token is not a valid JWT encoding")]
     JwtEncoding(#[from] JwtEncodingError),
 
-    #[error(transparent)]
-    InvalidClaims(#[from] JwtClaimsError),
+    #[error("invalid JWT token audience")]
+    InvalidJwtTokenAudience,
+
+    #[error("JWT token has expired")]
+    JwtTokenHasExpired,
+
+    #[error("JWT token is not yet ready to use")]
+    JwtTokenNotYetReadyToUse,
 
     #[error("invalid P256 key")]
     InvalidP256Key(jose_jwk::crypto::Error),
@@ -644,19 +644,6 @@ pub enum JwtEncodingError {
     InvalidCompactForm,
 }
 
-#[derive(Error, Debug, PartialEq)]
-#[non_exhaustive]
-pub enum JwtClaimsError {
-    #[error("invalid JWT token audience")]
-    InvalidJwtTokenAudience,
-
-    #[error("JWT token has expired")]
-    JwtTokenHasExpired,
-
-    #[error("JWT token is not yet ready to use")]
-    JwtTokenNotYetReadyToUse,
-}
-
 #[allow(dead_code, reason = "Debug use only")]
 #[derive(Debug)]
 pub(crate) enum KeyType {
@@ -693,8 +680,6 @@ mod tests {
     use hyper_util::rt::TokioIo;
     use rand::rngs::OsRng;
     use rsa::pkcs8::DecodePrivateKey;
-    use serde::Serialize;
-    use serde_json::json;
     use signature::Signer;
     use tokio::net::TcpListener;
 
@@ -708,7 +693,6 @@ mod tests {
             key: jose_jwk::Key::Ec(pk),
             prm: jose_jwk::Parameters {
                 kid: Some(kid),
-                alg: Some(jose_jwa::Algorithm::Signing(jose_jwa::Signing::Es256)),
                 ..Default::default()
             },
         };
@@ -722,47 +706,24 @@ mod tests {
             key: jose_jwk::Key::Rsa(pk),
             prm: jose_jwk::Parameters {
                 kid: Some(kid),
-                alg: Some(jose_jwa::Algorithm::Signing(jose_jwa::Signing::Rs256)),
                 ..Default::default()
             },
         };
         (sk, jwk)
     }
 
-    fn now() -> u64 {
-        SystemTime::now()
-            .duration_since(SystemTime::UNIX_EPOCH)
-            .unwrap()
-            .as_secs()
-    }
-
     fn build_jwt_payload(kid: String, sig: jose_jwa::Signing) -> String {
-        let now = now();
-        let body = typed_json::json! {{
-            "exp": now + 3600,
-            "nbf": now,
-            "aud": ["audience1", "neon", "audience2"],
-            "sub": "user1",
-            "sid": "session1",
-            "jti": "token1",
-            "iss": "neon-testing",
-        }};
-        build_custom_jwt_payload(kid, body, sig)
-    }
-
-    fn build_custom_jwt_payload(
-        kid: String,
-        body: impl Serialize,
-        sig: jose_jwa::Signing,
-    ) -> String {
         let header = JwtHeader {
             algorithm: jose_jwa::Algorithm::Signing(sig),
-            key_id: Some(Cow::Owned(kid)),
+            key_id: Some(&kid),
         };
+        let body = typed_json::json! {{
+            "exp": SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap().as_secs() + 3600,
+        }};
 
         let header =
             base64::encode_config(serde_json::to_string(&header).unwrap(), URL_SAFE_NO_PAD);
-        let body = base64::encode_config(serde_json::to_string(&body).unwrap(), URL_SAFE_NO_PAD);
+        let body = base64::encode_config(body.to_string(), URL_SAFE_NO_PAD);
 
         format!("{header}.{body}")
     }
@@ -777,16 +738,6 @@ mod tests {
         format!("{payload}.{sig}")
     }
 
-    fn new_custom_ec_jwt(kid: String, key: &p256::SecretKey, body: impl Serialize) -> String {
-        use p256::ecdsa::{Signature, SigningKey};
-
-        let payload = build_custom_jwt_payload(kid, body, jose_jwa::Signing::Es256);
-        let sig: Signature = SigningKey::from(key).sign(payload.as_bytes());
-        let sig = base64::encode_config(sig.to_bytes(), URL_SAFE_NO_PAD);
-
-        format!("{payload}.{sig}")
-    }
-
     fn new_rsa_jwt(kid: String, key: rsa::RsaPrivateKey) -> String {
         use rsa::pkcs1v15::SigningKey;
         use rsa::signature::SignatureEncoding;
@@ -858,34 +809,37 @@ X0n5X2/pBLJzxZc62ccvZYVnctBiFs6HbSnxpuMQCfkt/BcR/ttIepBQQIW86wHL
 -----END PRIVATE KEY-----
 ";
 
-    #[derive(Clone)]
-    struct Fetch(Vec<AuthRule>);
+    #[tokio::test]
+    async fn renew() {
+        let (rs1, jwk1) = new_rsa_jwk(RS1, "1".into());
+        let (rs2, jwk2) = new_rsa_jwk(RS2, "2".into());
+        let (ec1, jwk3) = new_ec_jwk("3".into());
+        let (ec2, jwk4) = new_ec_jwk("4".into());
 
-    impl FetchAuthRules for Fetch {
-        async fn fetch_auth_rules(
-            &self,
-            _ctx: &RequestMonitoring,
-            _endpoint: EndpointId,
-        ) -> Result<Vec<AuthRule>, FetchAuthRulesError> {
-            Ok(self.0.clone())
-        }
-    }
+        let foo_jwks = jose_jwk::JwkSet {
+            keys: vec![jwk1, jwk3],
+        };
+        let bar_jwks = jose_jwk::JwkSet {
+            keys: vec![jwk2, jwk4],
+        };
 
-    async fn jwks_server(
-        router: impl for<'a> Fn(&'a str) -> Option<Vec<u8>> + Send + Sync + 'static,
-    ) -> SocketAddr {
-        let router = Arc::new(router);
         let service = service_fn(move |req| {
-            let router = Arc::clone(&router);
+            let foo_jwks = foo_jwks.clone();
+            let bar_jwks = bar_jwks.clone();
             async move {
-                match router(req.uri().path()) {
-                    Some(body) => Response::builder()
-                        .status(200)
-                        .body(Full::new(Bytes::from(body))),
-                    None => Response::builder()
-                        .status(404)
-                        .body(Full::new(Bytes::new())),
-                }
+                let jwks = match req.uri().path() {
+                    "/foo" => &foo_jwks,
+                    "/bar" => &bar_jwks,
+                    _ => {
+                        return Response::builder()
+                            .status(404)
+                            .body(Full::new(Bytes::new()));
+                    }
+                };
+                let body = serde_json::to_vec(jwks).unwrap();
+                Response::builder()
+                    .status(200)
+                    .body(Full::new(Bytes::from(body)))
             }
         });
 
@@ -900,61 +854,84 @@ X0n5X2/pBLJzxZc62ccvZYVnctBiFs6HbSnxpuMQCfkt/BcR/ttIepBQQIW86wHL
             }
         });
 
-        addr
-    }
+        let client = reqwest::Client::new();
 
-    #[tokio::test]
-    async fn check_jwt_happy_path() {
-        let (rs1, jwk1) = new_rsa_jwk(RS1, "rs1".into());
-        let (rs2, jwk2) = new_rsa_jwk(RS2, "rs2".into());
-        let (ec1, jwk3) = new_ec_jwk("ec1".into());
-        let (ec2, jwk4) = new_ec_jwk("ec2".into());
+        #[derive(Clone)]
+        struct Fetch(SocketAddr, Vec<RoleNameInt>);
 
-        let foo_jwks = jose_jwk::JwkSet {
-            keys: vec![jwk1, jwk3],
-        };
-        let bar_jwks = jose_jwk::JwkSet {
-            keys: vec![jwk2, jwk4],
-        };
-
-        let jwks_addr = jwks_server(move |path| match path {
-            "/foo" => Some(serde_json::to_vec(&foo_jwks).unwrap()),
-            "/bar" => Some(serde_json::to_vec(&bar_jwks).unwrap()),
-            _ => None,
-        })
-        .await;
+        impl FetchAuthRules for Fetch {
+            async fn fetch_auth_rules(
+                &self,
+                _ctx: &RequestMonitoring,
+                _endpoint: EndpointId,
+            ) -> Result<Vec<AuthRule>, FetchAuthRulesError> {
+                Ok(vec![
+                    AuthRule {
+                        id: "foo".to_owned(),
+                        jwks_url: format!("http://{}/foo", self.0).parse().unwrap(),
+                        audience: None,
+                        role_names: self.1.clone(),
+                    },
+                    AuthRule {
+                        id: "bar".to_owned(),
+                        jwks_url: format!("http://{}/bar", self.0).parse().unwrap(),
+                        audience: None,
+                        role_names: self.1.clone(),
+                    },
+                ])
+            }
+        }
 
         let role_name1 = RoleName::from("anonymous");
         let role_name2 = RoleName::from("authenticated");
 
-        let roles = vec![
-            RoleNameInt::from(&role_name1),
-            RoleNameInt::from(&role_name2),
-        ];
-        let rules = vec![
-            AuthRule {
-                id: "foo".to_owned(),
-                jwks_url: format!("http://{jwks_addr}/foo").parse().unwrap(),
-                audience: None,
-                role_names: roles.clone(),
-            },
-            AuthRule {
-                id: "bar".to_owned(),
-                jwks_url: format!("http://{jwks_addr}/bar").parse().unwrap(),
-                audience: None,
-                role_names: roles.clone(),
-            },
-        ];
-
-        let fetch = Fetch(rules);
-        let jwk_cache = JwkCache::default();
+        let fetch = Fetch(
+            addr,
+            vec![
+                RoleNameInt::from(&role_name1),
+                RoleNameInt::from(&role_name2),
+            ],
+        );
 
         let endpoint = EndpointId::from("ep");
 
-        let jwt1 = new_rsa_jwt("rs1".into(), rs1);
-        let jwt2 = new_rsa_jwt("rs2".into(), rs2);
-        let jwt3 = new_ec_jwt("ec1".into(), &ec1);
-        let jwt4 = new_ec_jwt("ec2".into(), &ec2);
+        let jwk_cache = Arc::new(JwkCacheEntryLock::default());
+
+        let jwt1 = new_rsa_jwt("1".into(), rs1);
+        let jwt2 = new_rsa_jwt("2".into(), rs2);
+        let jwt3 = new_ec_jwt("3".into(), &ec1);
+        let jwt4 = new_ec_jwt("4".into(), &ec2);
+
+        // had the wrong kid, therefore will have the wrong ecdsa signature
+        let bad_jwt = new_ec_jwt("3".into(), &ec2);
+        // this role_name is not accepted
+        let bad_role_name = RoleName::from("cloud_admin");
+
+        let err = jwk_cache
+            .check_jwt(
+                &RequestMonitoring::test(),
+                &bad_jwt,
+                &client,
+                endpoint.clone(),
+                &role_name1,
+                &fetch,
+            )
+            .await
+            .unwrap_err();
+        assert!(err.to_string().contains("signature error"));
+
+        let err = jwk_cache
+            .check_jwt(
+                &RequestMonitoring::test(),
+                &jwt1,
+                &client,
+                endpoint.clone(),
+                &bad_role_name,
+                &fetch,
+            )
+            .await
+            .unwrap_err();
+        assert!(err.to_string().contains("jwk not found"));
 
         let tokens = [jwt1, jwt2, jwt3, jwt4];
         let role_names = [role_name1, role_name2];
@@ -963,250 +940,15 @@ X0n5X2/pBLJzxZc62ccvZYVnctBiFs6HbSnxpuMQCfkt/BcR/ttIepBQQIW86wHL
                 jwk_cache
                     .check_jwt(
                         &RequestMonitoring::test(),
+                        token,
+                        &client,
                         endpoint.clone(),
                         role,
                         &fetch,
-                        token,
                     )
                     .await
                     .unwrap();
             }
         }
     }
-
-    /// AWS Cognito escapes the `/` in the URL.
-    #[tokio::test]
-    async fn check_jwt_regression_cognito_issuer() {
-        let (key, jwk) = new_ec_jwk("key".into());
-
-        let now = now();
-        let token = new_custom_ec_jwt(
-            "key".into(),
-            &key,
-            typed_json::json! {{
-                "sub": "dd9a73fd-e785-4a13-aae1-e691ce43e89d",
-                // cognito uses `\/`. I cannot replicated that easily here as serde_json will refuse
-                // to write that escape character. instead I will make a bogus URL using `\` instead.
-                "iss": "https:\\\\cognito-idp.us-west-2.amazonaws.com\\us-west-2_abcdefgh",
-                "client_id": "abcdefghijklmnopqrstuvwxyz",
-                "origin_jti": "6759d132-3fe7-446e-9e90-2fe7e8017893",
-                "event_id": "ec9c36ab-b01d-46a0-94e4-87fde6767065",
-                "token_use": "access",
-                "scope": "aws.cognito.signin.user.admin",
-                "auth_time":now,
-                "exp":now + 60,
-                "iat":now,
-                "jti": "b241614b-0b93-4bdc-96db-0a3c7061d9c0",
-                "username": "dd9a73fd-e785-4a13-aae1-e691ce43e89d",
-            }},
-        );
-
-        let jwks = jose_jwk::JwkSet { keys: vec![jwk] };
-
-        let jwks_addr = jwks_server(move |_path| Some(serde_json::to_vec(&jwks).unwrap())).await;
-
-        let role_name = RoleName::from("anonymous");
-        let rules = vec![AuthRule {
-            id: "aws-cognito".to_owned(),
-            jwks_url: format!("http://{jwks_addr}/").parse().unwrap(),
-            audience: None,
-            role_names: vec![RoleNameInt::from(&role_name)],
-        }];
-
-        let fetch = Fetch(rules);
-        let jwk_cache = JwkCache::default();
-
-        let endpoint = EndpointId::from("ep");
-
-        jwk_cache
-            .check_jwt(
-                &RequestMonitoring::test(),
-                endpoint.clone(),
-                &role_name,
-                &fetch,
-                &token,
-            )
-            .await
-            .unwrap();
-    }
-
-    #[tokio::test]
-    async fn check_jwt_invalid_signature() {
-        let (_, jwk) = new_ec_jwk("1".into());
-        let (key, _) = new_ec_jwk("1".into());
-
-        // has a matching kid, but signed by the wrong key
-        let bad_jwt = new_ec_jwt("1".into(), &key);
-
-        let jwks = jose_jwk::JwkSet { keys: vec![jwk] };
-        let jwks_addr = jwks_server(move |path| match path {
-            "/" => Some(serde_json::to_vec(&jwks).unwrap()),
-            _ => None,
-        })
-        .await;
-
-        let role = RoleName::from("authenticated");
-
-        let rules = vec![AuthRule {
-            id: String::new(),
-            jwks_url: format!("http://{jwks_addr}/").parse().unwrap(),
-            audience: None,
-            role_names: vec![RoleNameInt::from(&role)],
-        }];
-
-        let fetch = Fetch(rules);
-        let jwk_cache = JwkCache::default();
-
-        let ep = EndpointId::from("ep");
-
-        let ctx = RequestMonitoring::test();
-        let err = jwk_cache
-            .check_jwt(&ctx, ep, &role, &fetch, &bad_jwt)
-            .await
-            .unwrap_err();
-        assert!(
-            matches!(err, JwtError::Signature(_)),
-            "expected \"signature error\", got {err:?}"
-        );
-    }
-
-    #[tokio::test]
-    async fn check_jwt_unknown_role() {
-        let (key, jwk) = new_rsa_jwk(RS1, "1".into());
-        let jwt = new_rsa_jwt("1".into(), key);
-
-        let jwks = jose_jwk::JwkSet { keys: vec![jwk] };
-        let jwks_addr = jwks_server(move |path| match path {
-            "/" => Some(serde_json::to_vec(&jwks).unwrap()),
-            _ => None,
-        })
-        .await;
-
-        let role = RoleName::from("authenticated");
-        let rules = vec![AuthRule {
-            id: String::new(),
-            jwks_url: format!("http://{jwks_addr}/").parse().unwrap(),
-            audience: None,
-            role_names: vec![RoleNameInt::from(&role)],
-        }];
-
-        let fetch = Fetch(rules);
-        let jwk_cache = JwkCache::default();
-
-        let ep = EndpointId::from("ep");
-
-        // this role_name is not accepted
-        let bad_role_name = RoleName::from("cloud_admin");
-
-        let ctx = RequestMonitoring::test();
-        let err = jwk_cache
-            .check_jwt(&ctx, ep, &bad_role_name, &fetch, &jwt)
-            .await
-            .unwrap_err();
-
-        assert!(
-            matches!(err, JwtError::JwkNotFound),
-            "expected \"jwk not found\", got {err:?}"
-        );
-    }
-
-    #[tokio::test]
-    async fn check_jwt_invalid_claims() {
-        let (key, jwk) = new_ec_jwk("1".into());
-
-        let jwks = jose_jwk::JwkSet { keys: vec![jwk] };
-        let jwks_addr = jwks_server(move |path| match path {
-            "/" => Some(serde_json::to_vec(&jwks).unwrap()),
-            _ => None,
-        })
-        .await;
-
-        let now = SystemTime::now()
-            .duration_since(SystemTime::UNIX_EPOCH)
-            .unwrap()
-            .as_secs();
-
-        struct Test {
-            body: serde_json::Value,
-            error: JwtClaimsError,
-        }
-
-        let table = vec![
-            Test {
-                body: json! {{
-                    "nbf": now + 60,
-                    "aud": "neon",
-                }},
-                error: JwtClaimsError::JwtTokenNotYetReadyToUse,
-            },
-            Test {
-                body: json! {{
-                    "exp": now - 60,
-                    "aud": ["neon"],
-                }},
-                error: JwtClaimsError::JwtTokenHasExpired,
-            },
-            Test {
-                body: json! {{
-                }},
-                error: JwtClaimsError::InvalidJwtTokenAudience,
-            },
-            Test {
-                body: json! {{
-                    "aud": [],
-                }},
-                error: JwtClaimsError::InvalidJwtTokenAudience,
-            },
-            Test {
-                body: json! {{
-                    "aud": "foo",
-                }},
-                error: JwtClaimsError::InvalidJwtTokenAudience,
-            },
-            Test {
-                body: json! {{
-                    "aud": ["foo"],
-                }},
-                error: JwtClaimsError::InvalidJwtTokenAudience,
-            },
-            Test {
-                body: json! {{
-                    "aud": ["foo", "bar"],
-                }},
-                error: JwtClaimsError::InvalidJwtTokenAudience,
-            },
-        ];
-
-        let role = RoleName::from("authenticated");
-
-        let rules = vec![AuthRule {
-            id: String::new(),
-            jwks_url: format!("http://{jwks_addr}/").parse().unwrap(),
-            audience: Some("neon".to_string()),
-            role_names: vec![RoleNameInt::from(&role)],
-        }];
-
-        let fetch = Fetch(rules);
-        let jwk_cache = JwkCache::default();
-
-        let ep = EndpointId::from("ep");
-
-        let ctx = RequestMonitoring::test();
-        for test in table {
-            let jwt = new_custom_ec_jwt("1".into(), &key, test.body);
-
-            match jwk_cache
-                .check_jwt(&ctx, ep.clone(), &role, &fetch, &jwt)
-                .await
-            {
-                Err(JwtError::InvalidClaims(error)) if error == test.error => {}
-                Err(err) => {
-                    panic!("expected {:?}, got {err:?}", test.error)
-                }
-                Ok(_payload) => {
-                    panic!("expected {:?}, got ok", test.error)
-                }
-            }
-        }
-    }
 }

storage_controller/src/service.rs

@@ -4958,7 +4958,16 @@ impl Service {
                     stripe_size,
                 },
                 placement_policy: Some(PlacementPolicy::Attached(0)), // No secondaries, for convenient debug/hacking
-                config: TenantConfig::default(),
+
+                // There is no way to know what the tenant's config was: revert to defaults
+                //
+                // TODO: remove `switch_aux_file_policy` once we finish auxv2 migration
+                //
+                // we write to both v1+v2 storage, so that the test case can use either storage format for testing
+                config: TenantConfig {
+                    switch_aux_file_policy: Some(models::AuxFilePolicy::CrossValidation),
+                    ..TenantConfig::default()
+                },
             })
             .await?;
 

test_runner/fixtures/neon_cli.py

@@ -16,6 +16,7 @@ from fixtures.common_types import Lsn, TenantId, TimelineId
 from fixtures.log_helper import log
 from fixtures.pageserver.common_types import IndexPartDump
 from fixtures.pg_version import PgVersion
+from fixtures.utils import AuxFileStore
 
 if TYPE_CHECKING:
     from typing import (
@@ -200,6 +201,7 @@ class NeonLocalCli(AbstractNeonCli):
         shard_stripe_size: Optional[int] = None,
         placement_policy: Optional[str] = None,
         set_default: bool = False,
+        aux_file_policy: Optional[AuxFileStore] = None,
     ):
         """
         Creates a new tenant, returns its id and its initial timeline's id.
@@ -221,6 +223,13 @@ class NeonLocalCli(AbstractNeonCli):
                 )
             )
 
+        if aux_file_policy is AuxFileStore.V2:
+            args.extend(["-c", "switch_aux_file_policy:v2"])
+        elif aux_file_policy is AuxFileStore.V1:
+            args.extend(["-c", "switch_aux_file_policy:v1"])
+        elif aux_file_policy is AuxFileStore.CrossValidation:
+            args.extend(["-c", "switch_aux_file_policy:cross-validation"])
+
         if set_default:
             args.append("--set-default")
 

test_runner/fixtures/neon_fixtures.py

@@ -40,6 +40,7 @@ from _pytest.fixtures import FixtureRequest
 from psycopg2.extensions import connection as PgConnection
 from psycopg2.extensions import cursor as PgCursor
 from psycopg2.extensions import make_dsn, parse_dsn
+from pytest_httpserver import HTTPServer
 from urllib3.util.retry import Retry
 
 from fixtures import overlayfs
@@ -94,6 +95,7 @@ from fixtures.utils import (
     subprocess_capture,
     wait_until,
 )
+from fixtures.utils import AuxFileStore as AuxFileStore  # reexport
 
 from .neon_api import NeonAPI, NeonApiEndpoint
 
@@ -352,6 +354,7 @@ class NeonEnvBuilder:
         initial_tenant: Optional[TenantId] = None,
         initial_timeline: Optional[TimelineId] = None,
         pageserver_virtual_file_io_engine: Optional[str] = None,
+        pageserver_aux_file_policy: Optional[AuxFileStore] = None,
         pageserver_default_tenant_config_compaction_algorithm: Optional[dict[str, Any]] = None,
         safekeeper_extra_opts: Optional[list[str]] = None,
         storage_controller_port_override: Optional[int] = None,
@@ -403,6 +406,8 @@ class NeonEnvBuilder:
                 f"Overriding pageserver default compaction algorithm to {self.pageserver_default_tenant_config_compaction_algorithm}"
             )
 
+        self.pageserver_aux_file_policy = pageserver_aux_file_policy
+
         self.safekeeper_extra_opts = safekeeper_extra_opts
 
         self.storage_controller_port_override = storage_controller_port_override
@@ -463,6 +468,7 @@ class NeonEnvBuilder:
             timeline_id=env.initial_timeline,
             shard_count=initial_tenant_shard_count,
             shard_stripe_size=initial_tenant_shard_stripe_size,
+            aux_file_policy=self.pageserver_aux_file_policy,
         )
         assert env.initial_tenant == initial_tenant
         assert env.initial_timeline == initial_timeline
@@ -1022,6 +1028,7 @@ class NeonEnv:
         self.control_plane_compute_hook_api = config.control_plane_compute_hook_api
 
         self.pageserver_virtual_file_io_engine = config.pageserver_virtual_file_io_engine
+        self.pageserver_aux_file_policy = config.pageserver_aux_file_policy
         self.pageserver_virtual_file_io_mode = config.pageserver_virtual_file_io_mode
 
         # Create the neon_local's `NeonLocalInitConf`
@@ -1317,6 +1324,7 @@ class NeonEnv:
         shard_stripe_size: Optional[int] = None,
         placement_policy: Optional[str] = None,
         set_default: bool = False,
+        aux_file_policy: Optional[AuxFileStore] = None,
     ) -> tuple[TenantId, TimelineId]:
         """
         Creates a new tenant, returns its id and its initial timeline's id.
@@ -1333,6 +1341,7 @@ class NeonEnv:
             shard_stripe_size=shard_stripe_size,
             placement_policy=placement_policy,
             set_default=set_default,
+            aux_file_policy=aux_file_policy,
         )
 
         return tenant_id, timeline_id
@@ -1390,6 +1399,7 @@ def neon_simple_env(
     compatibility_pg_distrib_dir: Path,
     pg_version: PgVersion,
     pageserver_virtual_file_io_engine: str,
+    pageserver_aux_file_policy: Optional[AuxFileStore],
     pageserver_default_tenant_config_compaction_algorithm: Optional[dict[str, Any]],
     pageserver_virtual_file_io_mode: Optional[str],
 ) -> Iterator[NeonEnv]:
@@ -1422,6 +1432,7 @@ def neon_simple_env(
         test_name=request.node.name,
         test_output_dir=test_output_dir,
         pageserver_virtual_file_io_engine=pageserver_virtual_file_io_engine,
+        pageserver_aux_file_policy=pageserver_aux_file_policy,
         pageserver_default_tenant_config_compaction_algorithm=pageserver_default_tenant_config_compaction_algorithm,
         pageserver_virtual_file_io_mode=pageserver_virtual_file_io_mode,
         combination=combination,
@@ -1448,6 +1459,7 @@ def neon_env_builder(
     top_output_dir: Path,
     pageserver_virtual_file_io_engine: str,
     pageserver_default_tenant_config_compaction_algorithm: Optional[dict[str, Any]],
+    pageserver_aux_file_policy: Optional[AuxFileStore],
     record_property: Callable[[str, object], None],
     pageserver_virtual_file_io_mode: Optional[str],
 ) -> Iterator[NeonEnvBuilder]:
@@ -1490,6 +1502,7 @@ def neon_env_builder(
         test_name=request.node.name,
         test_output_dir=test_output_dir,
         test_overlay_dir=test_overlay_dir,
+        pageserver_aux_file_policy=pageserver_aux_file_policy,
         pageserver_default_tenant_config_compaction_algorithm=pageserver_default_tenant_config_compaction_algorithm,
         pageserver_virtual_file_io_mode=pageserver_virtual_file_io_mode,
     ) as builder:
@@ -3086,10 +3099,6 @@ class NeonProxy(PgProtocol):
     class AuthBackend(abc.ABC):
         """All auth backends must inherit from this class"""
 
-        @property
-        def default_conn_url(self) -> Optional[str]:
-            return None
-
         @abc.abstractmethod
         def extra_args(self) -> list[str]:
             pass
@@ -3103,7 +3112,7 @@ class NeonProxy(PgProtocol):
                 *["--allow-self-signed-compute", "true"],
             ]
 
-    class Console(AuthBackend):
+    class ControlPlane(AuthBackend):
         def __init__(self, endpoint: str, fixed_rate_limit: Optional[int] = None):
             self.endpoint = endpoint
             self.fixed_rate_limit = fixed_rate_limit
@@ -3127,21 +3136,6 @@ class NeonProxy(PgProtocol):
                 ]
             return args
 
-    @dataclass(frozen=True)
-    class Postgres(AuthBackend):
-        pg_conn_url: str
-
-        @property
-        def default_conn_url(self) -> Optional[str]:
-            return self.pg_conn_url
-
-        def extra_args(self) -> list[str]:
-            return [
-                # Postgres auth backend params
-                *["--auth-backend", "postgres"],
-                *["--auth-endpoint", self.pg_conn_url],
-            ]
-
     def __init__(
         self,
         neon_binpath: Path,
@@ -3156,7 +3150,7 @@ class NeonProxy(PgProtocol):
     ):
         host = "127.0.0.1"
         domain = "proxy.localtest.me"  # resolves to 127.0.0.1
-        super().__init__(dsn=auth_backend.default_conn_url, host=domain, port=proxy_port)
+        super().__init__(host=domain, port=proxy_port)
 
         self.domain = domain
         self.host = host
@@ -3410,20 +3404,39 @@ def static_proxy(
     port_distributor: PortDistributor,
     neon_binpath: Path,
     test_output_dir: Path,
+    httpserver: HTTPServer,
 ) -> Iterator[NeonProxy]:
-    """Neon proxy that routes directly to vanilla postgres."""
+    """Neon proxy that routes directly to vanilla postgres and a mocked cplane HTTP API."""
 
     port = vanilla_pg.default_options["port"]
     host = vanilla_pg.default_options["host"]
     dbname = vanilla_pg.default_options["dbname"]
-    auth_endpoint = f"postgres://proxy:password@{host}:{port}/{dbname}"
 
-    # For simplicity, we use the same user for both `--auth-endpoint` and `safe_psql`
     vanilla_pg.start()
     vanilla_pg.safe_psql("create user proxy with login superuser password 'password'")
-    vanilla_pg.safe_psql("CREATE SCHEMA IF NOT EXISTS neon_control_plane")
-    vanilla_pg.safe_psql(
-        "CREATE TABLE neon_control_plane.endpoints (endpoint_id VARCHAR(255) PRIMARY KEY, allowed_ips VARCHAR(255))"
+    [(rolpassword,)] = vanilla_pg.safe_psql(
+        "select rolpassword from pg_catalog.pg_authid where rolname = 'proxy'"
+    )
+
+    # return local postgres addr on ProxyWakeCompute.
+    httpserver.expect_request("/cplane/proxy_wake_compute").respond_with_json(
+        {
+            "address": f"{host}:{port}",
+            "aux": {
+                "endpoint_id": "ep-foo-bar-1234",
+                "branch_id": "br-foo-bar",
+                "project_id": "foo-bar",
+            },
+        }
+    )
+
+    # return local postgres addr on ProxyWakeCompute.
+    httpserver.expect_request("/cplane/proxy_get_role_secret").respond_with_json(
+        {
+            "role_secret": rolpassword,
+            "allowed_ips": None,
+            "project_id": "foo-bar",
+        }
     )
 
     proxy_port = port_distributor.get_port()
@@ -3438,8 +3451,12 @@ def static_proxy(
         http_port=http_port,
         mgmt_port=mgmt_port,
         external_http_port=external_http_port,
-        auth_backend=NeonProxy.Postgres(auth_endpoint),
+        auth_backend=NeonProxy.ControlPlane(httpserver.url_for("/cplane")),
     ) as proxy:
+        proxy.default_options["user"] = "proxy"
+        proxy.default_options["password"] = "password"
+        proxy.default_options["dbname"] = dbname
+
         proxy.start()
         yield proxy
 

test_runner/fixtures/parametrize.py

@@ -10,6 +10,12 @@ from _pytest.python import Metafunc
 
 from fixtures.pg_version import PgVersion
 
+if TYPE_CHECKING:
+    from typing import Any, Optional
+
+    from fixtures.utils import AuxFileStore
+
+
 if TYPE_CHECKING:
     from typing import Any, Optional
 
@@ -44,6 +50,11 @@ def pageserver_virtual_file_io_mode() -> Optional[str]:
     return os.getenv("PAGESERVER_VIRTUAL_FILE_IO_MODE")
 
 
+@pytest.fixture(scope="function", autouse=True)
+def pageserver_aux_file_policy() -> Optional[AuxFileStore]:
+    return None
+
+
 def get_pageserver_default_tenant_config_compaction_algorithm() -> Optional[dict[str, Any]]:
     toml_table = os.getenv("PAGESERVER_DEFAULT_TENANT_CONFIG_COMPACTION_ALGORITHM")
     if toml_table is None:

test_runner/fixtures/utils.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import contextlib
+import enum
 import json
 import os
 import re
@@ -514,6 +515,21 @@ def assert_no_errors(log_file: Path, service: str, allowed_errors: list[str]):
     assert not errors, f"First log error on {service}: {errors[0]}\nHint: use scripts/check_allowed_errors.sh to test any new allowed_error you add"
 
 
+@enum.unique
+class AuxFileStore(str, enum.Enum):
+    V1 = "v1"
+    V2 = "v2"
+    CrossValidation = "cross-validation"
+
+    @override
+    def __repr__(self) -> str:
+        return f"'aux-{self.value}'"
+
+    @override
+    def __str__(self) -> str:
+        return f"'aux-{self.value}'"
+
+
 def assert_pageserver_backups_equal(left: Path, right: Path, skip_files: set[str]):
     """
     This is essentially:

test_runner/performance/test_logical_replication.py

@@ -9,7 +9,7 @@ import pytest
 from fixtures.benchmark_fixture import MetricReport
 from fixtures.common_types import Lsn
 from fixtures.log_helper import log
-from fixtures.neon_fixtures import logical_replication_sync
+from fixtures.neon_fixtures import AuxFileStore, logical_replication_sync
 
 if TYPE_CHECKING:
     from fixtures.benchmark_fixture import NeonBenchmarker
@@ -17,6 +17,7 @@ if TYPE_CHECKING:
     from fixtures.neon_fixtures import NeonEnv, PgBin
 
 
+@pytest.mark.parametrize("pageserver_aux_file_policy", [AuxFileStore.V2])
 @pytest.mark.timeout(1000)
 def test_logical_replication(neon_simple_env: NeonEnv, pg_bin: PgBin, vanilla_pg):
     env = neon_simple_env

test_runner/regress/test_attach_tenant_config.py

@@ -172,6 +172,7 @@ def test_fully_custom_config(positive_env: NeonEnv):
         },
         "walreceiver_connect_timeout": "13m",
         "image_layer_creation_check_threshold": 1,
+        "switch_aux_file_policy": "cross-validation",
         "lsn_lease_length": "1m",
         "lsn_lease_length_for_ts": "5s",
     }

test_runner/regress/test_ddl_forwarding.py

@@ -7,7 +7,6 @@ import psycopg2
 import pytest
 from fixtures.log_helper import log
 from fixtures.neon_fixtures import NeonEnv, VanillaPostgres
-from psycopg2.errors import UndefinedObject
 from pytest_httpserver import HTTPServer
 from werkzeug.wrappers.request import Request
 from werkzeug.wrappers.response import Response
@@ -336,34 +335,3 @@ def test_ddl_forwarding_invalid_db(neon_simple_env: NeonEnv):
         if not result:
             raise AssertionError("Could not count databases")
         assert result[0] == 0, "Database 'failure' still exists after restart"
-
-
-def test_ddl_forwarding_role_specs(neon_simple_env: NeonEnv):
-    """
-    Postgres has a concept of role specs:
-
-        ROLESPEC_CSTRING: ALTER ROLE xyz
-        ROLESPEC_CURRENT_USER: ALTER ROLE current_user
-        ROLESPEC_CURRENT_ROLE: ALTER ROLE current_role
-        ROLESPEC_SESSION_USER: ALTER ROLE session_user
-        ROLESPEC_PUBLIC: ALTER ROLE public
-
-    The extension is required to serialize these special role spec into
-    usernames for the purpose of DDL forwarding.
-    """
-    env = neon_simple_env
-
-    endpoint = env.endpoints.create_start("main")
-
-    with endpoint.cursor() as cur:
-        # ROLESPEC_CSTRING
-        cur.execute("ALTER ROLE cloud_admin WITH PASSWORD 'york'")
-        # ROLESPEC_CURRENT_USER
-        cur.execute("ALTER ROLE current_user WITH PASSWORD 'pork'")
-        # ROLESPEC_CURRENT_ROLE
-        cur.execute("ALTER ROLE current_role WITH PASSWORD 'cork'")
-        # ROLESPEC_SESSION_USER
-        cur.execute("ALTER ROLE session_user WITH PASSWORD 'bork'")
-        # ROLESPEC_PUBLIC
-        with pytest.raises(UndefinedObject):
-            cur.execute("ALTER ROLE public WITH PASSWORD 'dork'")

test_runner/regress/test_logical_replication.py

@@ -5,9 +5,11 @@ from functools import partial
 from random import choice
 from string import ascii_lowercase
 
+import pytest
 from fixtures.common_types import Lsn
 from fixtures.log_helper import log
 from fixtures.neon_fixtures import (
+    AuxFileStore,
     NeonEnv,
     NeonEnvBuilder,
     PgProtocol,
@@ -21,6 +23,17 @@ def random_string(n: int):
     return "".join([choice(ascii_lowercase) for _ in range(n)])
 
 
+@pytest.mark.parametrize(
+    "pageserver_aux_file_policy", [AuxFileStore.V2, AuxFileStore.CrossValidation]
+)
+def test_aux_file_v2_flag(neon_simple_env: NeonEnv, pageserver_aux_file_policy: AuxFileStore):
+    env = neon_simple_env
+    with env.pageserver.http_client() as client:
+        tenant_config = client.tenant_config(env.initial_tenant).effective_config
+        assert pageserver_aux_file_policy == tenant_config["switch_aux_file_policy"]
+
+
+@pytest.mark.parametrize("pageserver_aux_file_policy", [AuxFileStore.CrossValidation])
 def test_logical_replication(neon_simple_env: NeonEnv, vanilla_pg):
     env = neon_simple_env
 
@@ -160,6 +173,7 @@ COMMIT;
 
 
 # Test that neon.logical_replication_max_snap_files works
+@pytest.mark.parametrize("pageserver_aux_file_policy", [AuxFileStore.CrossValidation])
 def test_obsolete_slot_drop(neon_simple_env: NeonEnv, vanilla_pg):
     def slot_removed(ep):
         assert (
@@ -336,6 +350,7 @@ FROM generate_series(1, 16384) AS seq; -- Inserts enough rows to exceed 16MB of
 #
 # Most pages start with a contrecord, so we don't do anything special
 # to ensure that.
+@pytest.mark.parametrize("pageserver_aux_file_policy", [AuxFileStore.CrossValidation])
 def test_restart_endpoint(neon_simple_env: NeonEnv, vanilla_pg):
     env = neon_simple_env
 
@@ -380,6 +395,7 @@ def test_restart_endpoint(neon_simple_env: NeonEnv, vanilla_pg):
 # logical replication bug as such, but without logical replication,
 # records passed ot the WAL redo process are never large enough to hit
 # the bug.
+@pytest.mark.parametrize("pageserver_aux_file_policy", [AuxFileStore.CrossValidation])
 def test_large_records(neon_simple_env: NeonEnv, vanilla_pg):
     env = neon_simple_env
 
@@ -451,6 +467,7 @@ def test_slots_and_branching(neon_simple_env: NeonEnv):
     ws_cur.execute("select pg_create_logical_replication_slot('my_slot', 'pgoutput')")
 
 
+@pytest.mark.parametrize("pageserver_aux_file_policy", [AuxFileStore.CrossValidation])
 def test_replication_shutdown(neon_simple_env: NeonEnv):
     # Ensure Postgres can exit without stuck when a replication job is active + neon extension installed
     env = neon_simple_env

test_runner/regress/test_proxy_allowed_ips.py

@@ -6,20 +6,27 @@ from fixtures.neon_fixtures import (
     NeonProxy,
     VanillaPostgres,
 )
+from pytest_httpserver import HTTPServer
 
 TABLE_NAME = "neon_control_plane.endpoints"
 
 
-# Proxy uses the same logic for psql and websockets.
-@pytest.mark.asyncio
-async def test_proxy_psql_allowed_ips(static_proxy: NeonProxy, vanilla_pg: VanillaPostgres):
-    # Shouldn't be able to connect to this project
-    vanilla_pg.safe_psql(
-        f"INSERT INTO {TABLE_NAME} (endpoint_id, allowed_ips) VALUES ('private-project', '8.8.8.8')"
+def test_proxy_psql_not_allowed_ips(
+    static_proxy: NeonProxy,
+    vanilla_pg: VanillaPostgres,
+    httpserver: HTTPServer,
+):
+    [(rolpassword,)] = vanilla_pg.safe_psql(
+        "select rolpassword from pg_catalog.pg_authid where rolname = 'proxy'"
     )
-    # Should be able to connect to this project
-    vanilla_pg.safe_psql(
-        f"INSERT INTO {TABLE_NAME} (endpoint_id, allowed_ips) VALUES ('generic-project', '::1,127.0.0.1')"
+
+    # Shouldn't be able to connect to this project
+    httpserver.expect_request("/cplane/proxy_get_role_secret").respond_with_json(
+        {
+            "role_secret": rolpassword,
+            "allowed_ips": ["8.8.8.8"],
+            "project_id": "foo-bar",
+        }
     )
 
     def check_cannot_connect(**kwargs):
@@ -37,6 +44,25 @@ async def test_proxy_psql_allowed_ips(static_proxy: NeonProxy, vanilla_pg: Vanil
     # with SNI
     check_cannot_connect(query="select 1", host="private-project.localtest.me")
 
+
+def test_proxy_psql_allowed_ips(
+    static_proxy: NeonProxy,
+    vanilla_pg: VanillaPostgres,
+    httpserver: HTTPServer,
+):
+    [(rolpassword,)] = vanilla_pg.safe_psql(
+        "select rolpassword from pg_catalog.pg_authid where rolname = 'proxy'"
+    )
+
+    # Should be able to connect to this project
+    httpserver.expect_request("/cplane/proxy_get_role_secret").respond_with_json(
+        {
+            "role_secret": rolpassword,
+            "allowed_ips": ["::1", "127.0.0.1"],
+            "project_id": "foo-bar",
+        }
+    )
+
     # no SNI, deprecated `options=project` syntax (before we had several endpoint in project)
     out = static_proxy.safe_psql(query="select 1", sslsni=0, options="project=generic-project")
     assert out[0][0] == 1
@@ -50,27 +76,61 @@ async def test_proxy_psql_allowed_ips(static_proxy: NeonProxy, vanilla_pg: Vanil
     assert out[0][0] == 1
 
 
-@pytest.mark.asyncio
-async def test_proxy_http_allowed_ips(static_proxy: NeonProxy, vanilla_pg: VanillaPostgres):
-    static_proxy.safe_psql("create user http_auth with password 'http' superuser")
+def test_proxy_http_not_allowed_ips(
+    static_proxy: NeonProxy,
+    vanilla_pg: VanillaPostgres,
+    httpserver: HTTPServer,
+):
+    vanilla_pg.safe_psql("create user http_auth with password 'http' superuser")
 
-    # Shouldn't be able to connect to this project
-    vanilla_pg.safe_psql(
-        f"INSERT INTO {TABLE_NAME} (endpoint_id, allowed_ips) VALUES ('proxy', '8.8.8.8')"
+    [(rolpassword,)] = vanilla_pg.safe_psql(
+        "select rolpassword from pg_catalog.pg_authid where rolname = 'http_auth'"
     )
 
-    def query(status: int, query: str, *args):
+    httpserver.expect_oneshot_request("/cplane/proxy_get_role_secret").respond_with_json(
+        {
+            "role_secret": rolpassword,
+            "allowed_ips": ["8.8.8.8"],
+            "project_id": "foo-bar",
+        }
+    )
+
+    with httpserver.wait() as waiting:
         static_proxy.http_query(
-            query,
-            args,
+            "select 1;",
+            [],
             user="http_auth",
             password="http",
-            expected_code=status,
+            expected_code=400,
         )
+    assert waiting.result
 
-    query(400, "select 1;")  # ip address is not allowed
-    # Should be able to connect to this project
-    vanilla_pg.safe_psql(
-        f"UPDATE {TABLE_NAME} SET allowed_ips = '8.8.8.8,127.0.0.1' WHERE endpoint_id = 'proxy'"
+
+def test_proxy_http_allowed_ips(
+    static_proxy: NeonProxy,
+    vanilla_pg: VanillaPostgres,
+    httpserver: HTTPServer,
+):
+    vanilla_pg.safe_psql("create user http_auth with password 'http' superuser")
+
+    [(rolpassword,)] = vanilla_pg.safe_psql(
+        "select rolpassword from pg_catalog.pg_authid where rolname = 'http_auth'"
     )
-    query(200, "select 1;")  # should work now
+
+    httpserver.expect_oneshot_request("/cplane/proxy_get_role_secret").respond_with_json(
+        {
+            "role_secret": rolpassword,
+            "allowed_ips": ["8.8.8.8", "127.0.0.1"],
+            "project_id": "foo-bar",
+        }
+    )
+
+    with httpserver.wait() as waiting:
+        static_proxy.http_query(
+            "select 1;",
+            [],
+            user="http_auth",
+            password="http",
+            expected_code=200,
+        )
+    assert waiting.result

test_runner/regress/test_timeline_archive.py

@@ -137,17 +137,14 @@ def test_timeline_offloading(neon_env_builder: NeonEnvBuilder, manual_offload: b
         }
     )
 
-    # Create three branches that depend on each other, starting with two
-    grandparent_timeline_id = env.create_branch(
-        "test_ancestor_branch_archive_grandparent", tenant_id
-    )
-    parent_timeline_id = env.create_branch(
-        "test_ancestor_branch_archive_parent", tenant_id, "test_ancestor_branch_archive_grandparent"
+    # Create two branches and archive them
+    parent_timeline_id = env.create_branch("test_ancestor_branch_archive_parent", tenant_id)
+    leaf_timeline_id = env.create_branch(
+        "test_ancestor_branch_archive_branch1", tenant_id, "test_ancestor_branch_archive_parent"
     )
 
-    # write some stuff to the parent
     with env.endpoints.create_start(
-        "test_ancestor_branch_archive_parent", tenant_id=tenant_id
+        "test_ancestor_branch_archive_branch1", tenant_id=tenant_id
     ) as endpoint:
         endpoint.safe_psql_many(
             [
@@ -157,11 +154,6 @@ def test_timeline_offloading(neon_env_builder: NeonEnvBuilder, manual_offload: b
         )
         sum = endpoint.safe_psql("SELECT sum(key) from foo where key > 50")
 
-    # create the third branch
-    leaf_timeline_id = env.create_branch(
-        "test_ancestor_branch_archive_branch1", tenant_id, "test_ancestor_branch_archive_parent"
-    )
-
     ps_http.timeline_archival_config(
         tenant_id,
         leaf_timeline_id,
@@ -179,12 +171,6 @@ def test_timeline_offloading(neon_env_builder: NeonEnvBuilder, manual_offload: b
         state=TimelineArchivalState.ARCHIVED,
     )
 
-    ps_http.timeline_archival_config(
-        tenant_id,
-        grandparent_timeline_id,
-        state=TimelineArchivalState.ARCHIVED,
-    )
-
     def timeline_offloaded_logged(timeline_id: TimelineId) -> bool:
         return (
             env.pageserver.log_contains(f".*{timeline_id}.* offloading archived timeline.*")
@@ -215,34 +201,30 @@ def test_timeline_offloading(neon_env_builder: NeonEnvBuilder, manual_offload: b
 
     ps_http.timeline_archival_config(
         tenant_id,
-        grandparent_timeline_id,
+        parent_timeline_id,
         state=TimelineArchivalState.UNARCHIVED,
     )
     ps_http.timeline_archival_config(
         tenant_id,
-        parent_timeline_id,
+        leaf_timeline_id,
         state=TimelineArchivalState.UNARCHIVED,
     )
-    parent_detail = ps_http.timeline_detail(
+    leaf_detail = ps_http.timeline_detail(
         tenant_id,
-        parent_timeline_id,
+        leaf_timeline_id,
     )
-    assert parent_detail["is_archived"] is False
+    assert leaf_detail["is_archived"] is False
 
     with env.endpoints.create_start(
-        "test_ancestor_branch_archive_parent", tenant_id=tenant_id
+        "test_ancestor_branch_archive_branch1", tenant_id=tenant_id
     ) as endpoint:
         sum_again = endpoint.safe_psql("SELECT sum(key) from foo where key > 50")
         assert sum == sum_again
 
-    # Test that deletion of offloaded timelines works
-    ps_http.timeline_delete(tenant_id, leaf_timeline_id)
-
     assert not timeline_offloaded_logged(initial_timeline_id)
 
 
-@pytest.mark.parametrize("delete_timeline", [False, True])
-def test_timeline_offload_persist(neon_env_builder: NeonEnvBuilder, delete_timeline: bool):
+def test_timeline_offload_persist(neon_env_builder: NeonEnvBuilder):
     """
     Test for persistence of timeline offload state
     """
@@ -324,35 +306,27 @@ def test_timeline_offload_persist(neon_env_builder: NeonEnvBuilder, delete_timel
     assert timeline_offloaded_api(child_timeline_id)
     assert not timeline_offloaded_api(root_timeline_id)
 
-    if delete_timeline:
-        ps_http.timeline_delete(tenant_id, child_timeline_id)
-        with pytest.raises(PageserverApiException, match="not found"):
-            ps_http.timeline_detail(
-                tenant_id,
-                child_timeline_id,
-            )
-    else:
-        ps_http.timeline_archival_config(
-            tenant_id,
-            child_timeline_id,
-            state=TimelineArchivalState.UNARCHIVED,
-        )
-        child_detail = ps_http.timeline_detail(
-            tenant_id,
-            child_timeline_id,
-        )
-        assert child_detail["is_archived"] is False
+    ps_http.timeline_archival_config(
+        tenant_id,
+        child_timeline_id,
+        state=TimelineArchivalState.UNARCHIVED,
+    )
+    child_detail = ps_http.timeline_detail(
+        tenant_id,
+        child_timeline_id,
+    )
+    assert child_detail["is_archived"] is False
 
-        with env.endpoints.create_start(
-            "test_archived_branch_persisted", tenant_id=tenant_id
-        ) as endpoint:
-            sum_again = endpoint.safe_psql("SELECT sum(key) from foo where key < 500")
-            assert sum == sum_again
+    with env.endpoints.create_start(
+        "test_archived_branch_persisted", tenant_id=tenant_id
+    ) as endpoint:
+        sum_again = endpoint.safe_psql("SELECT sum(key) from foo where key < 500")
+        assert sum == sum_again
 
-        assert_prefix_empty(
-            neon_env_builder.pageserver_remote_storage,
-            prefix=f"tenants/{str(env.initial_tenant)}/tenant-manifest",
-        )
+    assert_prefix_empty(
+        neon_env_builder.pageserver_remote_storage,
+        prefix=f"tenants/{str(env.initial_tenant)}/tenant-manifest",
+    )
 
     assert not timeline_offloaded_api(root_timeline_id)
 

test_runner/regress/test_unstable_extensions.py

@@ -1,50 +0,0 @@
-from __future__ import annotations
-
-from typing import TYPE_CHECKING, cast
-
-import pytest
-from psycopg2.errors import InsufficientPrivilege
-
-if TYPE_CHECKING:
-    from fixtures.neon_fixtures import NeonEnv
-
-
-def test_unstable_extensions_installation(neon_simple_env: NeonEnv):
-    """
-    Test that the unstable extension support within the neon extension can
-    block extension installation.
-    """
-    env = neon_simple_env
-
-    neon_unstable_extensions = "pg_prewarm,amcheck"
-
-    endpoint = env.endpoints.create(
-        "main",
-        config_lines=[
-            "neon.allow_unstable_extensions=false",
-            f"neon.unstable_extensions='{neon_unstable_extensions}'",
-        ],
-    )
-    endpoint.respec(skip_pg_catalog_updates=False)
-    endpoint.start()
-
-    with endpoint.cursor() as cursor:
-        cursor.execute("SELECT current_setting('neon.unstable_extensions')")
-        result = cursor.fetchone()
-        assert result is not None
-        setting = cast("str", result[0])
-        assert setting == neon_unstable_extensions
-
-        with pytest.raises(InsufficientPrivilege):
-            cursor.execute("CREATE EXTENSION pg_prewarm")
-
-        with pytest.raises(InsufficientPrivilege):
-            cursor.execute("CREATE EXTENSION amcheck")
-
-        # Make sure that we can install a "stable" extension
-        cursor.execute("CREATE EXTENSION pageinspect")
-
-        cursor.execute("BEGIN")
-        cursor.execute("SET neon.allow_unstable_extensions TO true")
-        cursor.execute("CREATE EXTENSION pg_prewarm")
-        cursor.execute("COMMIT")