From 8f0743b5b5fcd208618e21b05523672a33b6d9cb Mon Sep 17 00:00:00 2001
From: Alex Chi Z <chi@neon.tech>
Date: Mon, 13 May 2024 15:01:34 -0400
Subject: [PATCH 01/20] feat(pageserver): persist aux file policy in timeline
 metadata

Signed-off-by: Alex Chi Z <chi@neon.tech>
---
 control_plane/src/pageserver.rs               |   6 +-
 libs/pageserver_api/src/models.rs             | 100 +++++++++++++++++-
 pageserver/ctl/src/main.rs                    |   1 +
 pageserver/src/http/routes.rs                 |   2 +
 pageserver/src/pgdatadir_mapping.rs           |  49 +++++++--
 pageserver/src/tenant.rs                      |  16 +++
 pageserver/src/tenant/config.rs               |   8 +-
 .../src/tenant/remote_timeline_client.rs      |  15 ++-
 .../tenant/remote_timeline_client/index.rs    |  82 +++++++++++++-
 pageserver/src/tenant/timeline.rs             |  18 +++-
 pageserver/src/tenant/timeline/delete.rs      |   6 +-
 pageserver/src/tenant/upload_queue.rs         |   5 +
 test_runner/regress/test_aux_files.py         |  81 ++++++++++++++
 13 files changed, 361 insertions(+), 28 deletions(-)
 create mode 100644 test_runner/regress/test_aux_files.py

diff --git a/control_plane/src/pageserver.rs b/control_plane/src/pageserver.rs
index 5a8476369769..a8fa8486e8c5 100644
--- a/control_plane/src/pageserver.rs
+++ b/control_plane/src/pageserver.rs
@@ -17,7 +17,7 @@ use anyhow::{bail, Context};
 use camino::Utf8PathBuf;
 use futures::SinkExt;
 use pageserver_api::models::{
-    self, AuxFilePolicy, LocationConfig, ShardParameters, TenantHistorySize, TenantInfo,
+    self, LocationConfig, ShardParameters, SwitchAuxFilePolicy, TenantHistorySize, TenantInfo,
     TimelineInfo,
 };
 use pageserver_api::shard::TenantShardId;
@@ -380,7 +380,7 @@ impl PageServerNode {
                 .context("parse `timeline_get_throttle` from json")?,
             switch_aux_file_policy: settings
                 .remove("switch_aux_file_policy")
-                .map(|x| x.parse::<AuxFilePolicy>())
+                .map(|x| x.parse::<SwitchAuxFilePolicy>())
                 .transpose()
                 .context("Failed to parse 'switch_aux_file_policy'")?,
         };
@@ -503,7 +503,7 @@ impl PageServerNode {
                     .context("parse `timeline_get_throttle` from json")?,
                 switch_aux_file_policy: settings
                     .remove("switch_aux_file_policy")
-                    .map(|x| x.parse::<AuxFilePolicy>())
+                    .map(|x| x.parse::<SwitchAuxFilePolicy>())
                     .transpose()
                     .context("Failed to parse 'switch_aux_file_policy'")?,
             }
diff --git a/libs/pageserver_api/src/models.rs b/libs/pageserver_api/src/models.rs
index 1df5820fb946..6771bd9e19a8 100644
--- a/libs/pageserver_api/src/models.rs
+++ b/libs/pageserver_api/src/models.rs
@@ -10,6 +10,7 @@ use std::{
     io::{BufRead, Read},
     num::{NonZeroU64, NonZeroUsize},
     str::FromStr,
+    sync::atomic::AtomicUsize,
     time::{Duration, SystemTime},
 };
 
@@ -305,17 +306,107 @@ 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 switch_aux_file_policy: Option<SwitchAuxFilePolicy>,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
-pub enum AuxFilePolicy {
+pub enum SwitchAuxFilePolicy {
     V1,
     V2,
     CrossValidation,
 }
 
-impl FromStr for AuxFilePolicy {
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+pub enum RuntimeAuxFilePolicy {
+    V1,
+    V2,
+    CrossValidation,
+    Unspecified,
+}
+
+pub struct AtomicRuntimeAuxFilePolicy(AtomicUsize);
+
+impl AtomicRuntimeAuxFilePolicy {
+    pub fn new(policy: RuntimeAuxFilePolicy) -> Self {
+        Self(AtomicUsize::new(policy.to_usize()))
+    }
+
+    pub fn load(&self) -> RuntimeAuxFilePolicy {
+        RuntimeAuxFilePolicy::from_usize(self.0.load(std::sync::atomic::Ordering::SeqCst))
+    }
+
+    pub fn store(&self, policy: RuntimeAuxFilePolicy) {
+        self.0.store(
+            RuntimeAuxFilePolicy::to_usize(policy),
+            std::sync::atomic::Ordering::SeqCst,
+        );
+    }
+}
+
+impl SwitchAuxFilePolicy {
+    pub fn to_usize(self) -> usize {
+        match self {
+            Self::V1 => 1,
+            Self::CrossValidation => 2,
+            Self::V2 => 3,
+        }
+    }
+
+    pub fn from_usize(this: usize) -> Self {
+        match this {
+            1 => Self::V1,
+            2 => Self::CrossValidation,
+            3 => Self::V2,
+            _ => unreachable!(),
+        }
+    }
+
+    pub fn to_runtime_policy(&self) -> RuntimeAuxFilePolicy {
+        RuntimeAuxFilePolicy::from_usize(self.to_usize())
+    }
+}
+
+impl RuntimeAuxFilePolicy {
+    pub fn to_usize(self) -> usize {
+        match self {
+            Self::V1 => 1,
+            Self::CrossValidation => 2,
+            Self::V2 => 3,
+            Self::Unspecified => 0,
+        }
+    }
+
+    pub fn from_usize(this: usize) -> Self {
+        match this {
+            1 => Self::V1,
+            2 => Self::CrossValidation,
+            3 => Self::V2,
+            0 => Self::Unspecified,
+            _ => unreachable!(),
+        }
+    }
+}
+
+impl FromStr for RuntimeAuxFilePolicy {
+    type Err = anyhow::Error;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        let s = s.to_lowercase();
+        if s.is_empty() || s == "unspecified" {
+            Ok(Self::Unspecified)
+        } else if s == "v1" {
+            Ok(Self::V1)
+        } else if s == "v2" {
+            Ok(Self::V2)
+        } else if s == "crossvalidation" || s == "cross_validation" {
+            Ok(Self::CrossValidation)
+        } else {
+            anyhow::bail!("cannot parse {} to aux file policy", s)
+        }
+    }
+}
+
+impl FromStr for SwitchAuxFilePolicy {
     type Err = anyhow::Error;
 
     fn from_str(s: &str) -> Result<Self, Self::Err> {
@@ -604,6 +695,9 @@ pub struct TimelineInfo {
     pub state: TimelineState,
 
     pub walreceiver_status: String,
+
+    /// Whether aux file v2 is enabled
+    pub last_aux_file_policy: RuntimeAuxFilePolicy,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
diff --git a/pageserver/ctl/src/main.rs b/pageserver/ctl/src/main.rs
index 1fb75584fc92..e92c352dab1d 100644
--- a/pageserver/ctl/src/main.rs
+++ b/pageserver/ctl/src/main.rs
@@ -219,6 +219,7 @@ fn handle_metadata(
     let mut meta = TimelineMetadata::from_bytes(&metadata_bytes)?;
     println!("Current metadata:\n{meta:?}");
     let mut update_meta = false;
+    // TODO: simplify this part
     if let Some(disk_consistent_lsn) = disk_consistent_lsn {
         meta = TimelineMetadata::new(
             *disk_consistent_lsn,
diff --git a/pageserver/src/http/routes.rs b/pageserver/src/http/routes.rs
index a8ca642dc59a..8fcf84f5d1de 100644
--- a/pageserver/src/http/routes.rs
+++ b/pageserver/src/http/routes.rs
@@ -433,6 +433,8 @@ async fn build_timeline_info_common(
         state,
 
         walreceiver_status,
+
+        last_aux_file_policy: timeline.last_aux_file_policy.load(),
     };
     Ok(info)
 }
diff --git a/pageserver/src/pgdatadir_mapping.rs b/pageserver/src/pgdatadir_mapping.rs
index 1092d64d33fe..3bd06eee35dd 100644
--- a/pageserver/src/pgdatadir_mapping.rs
+++ b/pageserver/src/pgdatadir_mapping.rs
@@ -24,7 +24,7 @@ use pageserver_api::key::{
     AUX_FILES_KEY, CHECKPOINT_KEY, CONTROLFILE_KEY, DBDIR_KEY, TWOPHASEDIR_KEY,
 };
 use pageserver_api::keyspace::SparseKeySpace;
-use pageserver_api::models::AuxFilePolicy;
+use pageserver_api::models::{RuntimeAuxFilePolicy, SwitchAuxFilePolicy};
 use pageserver_api::reltag::{BlockNumber, RelTag, SlruKind};
 use postgres_ffi::relfile_utils::{FSM_FORKNUM, VISIBILITYMAP_FORKNUM};
 use postgres_ffi::BLCKSZ;
@@ -35,7 +35,7 @@ use std::ops::ControlFlow;
 use std::ops::Range;
 use strum::IntoEnumIterator;
 use tokio_util::sync::CancellationToken;
-use tracing::{debug, trace, warn};
+use tracing::{debug, info, trace, warn};
 use utils::bin_ser::DeserializeError;
 use utils::vec_map::{VecMap, VecMapOrdering};
 use utils::{bin_ser::BeSer, lsn::Lsn};
@@ -718,10 +718,13 @@ impl Timeline {
         lsn: Lsn,
         ctx: &RequestContext,
     ) -> Result<HashMap<String, Bytes>, PageReconstructError> {
-        match self.get_switch_aux_file_policy() {
-            AuxFilePolicy::V1 => self.list_aux_files_v1(lsn, ctx).await,
-            AuxFilePolicy::V2 => self.list_aux_files_v2(lsn, ctx).await,
-            AuxFilePolicy::CrossValidation => {
+        let current_policy = self.last_aux_file_policy.load();
+        match current_policy {
+            RuntimeAuxFilePolicy::V1 | RuntimeAuxFilePolicy::Unspecified => {
+                self.list_aux_files_v1(lsn, ctx).await
+            }
+            RuntimeAuxFilePolicy::V2 => self.list_aux_files_v2(lsn, ctx).await,
+            RuntimeAuxFilePolicy::CrossValidation => {
                 let v1_result = self.list_aux_files_v1(lsn, ctx).await;
                 let v2_result = self.list_aux_files_v2(lsn, ctx).await;
                 match (v1_result, v2_result) {
@@ -1451,7 +1454,7 @@ impl<'a> DatadirModification<'a> {
     }
 
     pub fn init_aux_dir(&mut self) -> anyhow::Result<()> {
-        if let AuxFilePolicy::V2 = self.tline.get_switch_aux_file_policy() {
+        if let SwitchAuxFilePolicy::V2 = self.tline.get_switch_aux_file_policy() {
             return Ok(());
         }
         let buf = AuxFilesDirectory::ser(&AuxFilesDirectory {
@@ -1469,8 +1472,34 @@ impl<'a> DatadirModification<'a> {
         content: &[u8],
         ctx: &RequestContext,
     ) -> anyhow::Result<()> {
-        let policy = self.tline.get_switch_aux_file_policy();
-        if let AuxFilePolicy::V2 | AuxFilePolicy::CrossValidation = policy {
+        let switch_policy = self.tline.get_switch_aux_file_policy();
+
+        let policy = {
+            let current_policy = self.tline.last_aux_file_policy.load();
+            // Allowed switch path:
+            // * no aux files -> v1/v2/cross-validation
+            // * cross-validation->v2
+            match (current_policy, switch_policy) {
+                (RuntimeAuxFilePolicy::Unspecified, _)
+                | (RuntimeAuxFilePolicy::CrossValidation, SwitchAuxFilePolicy::V2) => {
+                    let new_policy = switch_policy.to_runtime_policy();
+                    self.tline.last_aux_file_policy.store(new_policy);
+                    info!("switching to aux file policy {:?}", new_policy);
+                    if let Some(ref remote_client) = self.tline.remote_client {
+                        remote_client
+                            .schedule_index_upload_for_aux_file_policy_update(new_policy)?;
+                        remote_client.wait_completion().await?;
+                    }
+                    info!("finish switching to aux file policy {:?}", new_policy);
+                    new_policy
+                }
+                (current_policy, _) => current_policy,
+            }
+        };
+
+        assert!(policy != RuntimeAuxFilePolicy::Unspecified);
+
+        if let RuntimeAuxFilePolicy::V2 | RuntimeAuxFilePolicy::CrossValidation = policy {
             let key = aux_file::encode_aux_file_key(path);
             // retrieve the key from the engine
             let old_val = match self.get(key, ctx).await {
@@ -1521,7 +1550,7 @@ impl<'a> DatadirModification<'a> {
             self.put(key, Value::Image(new_val.into()));
         }
 
-        if let AuxFilePolicy::V1 | AuxFilePolicy::CrossValidation = policy {
+        if let RuntimeAuxFilePolicy::V1 | RuntimeAuxFilePolicy::CrossValidation = policy {
             let file_path = path.to_string();
             let content = if content.is_empty() {
                 None
diff --git a/pageserver/src/tenant.rs b/pageserver/src/tenant.rs
index 010e56a899db..bfe8ea57688b 100644
--- a/pageserver/src/tenant.rs
+++ b/pageserver/src/tenant.rs
@@ -20,6 +20,7 @@ use futures::stream::FuturesUnordered;
 use futures::FutureExt;
 use futures::StreamExt;
 use pageserver_api::models;
+use pageserver_api::models::RuntimeAuxFilePolicy;
 use pageserver_api::models::TimelineState;
 use pageserver_api::models::WalRedoManagerStatus;
 use pageserver_api::shard::ShardIdentity;
@@ -528,6 +529,7 @@ impl Tenant {
         index_part: Option<IndexPart>,
         metadata: TimelineMetadata,
         ancestor: Option<Arc<Timeline>>,
+        last_aux_file_policy: RuntimeAuxFilePolicy,
         _ctx: &RequestContext,
     ) -> anyhow::Result<()> {
         let tenant_id = self.tenant_shard_id;
@@ -538,6 +540,10 @@ impl Tenant {
             ancestor.clone(),
             resources,
             CreateTimelineCause::Load,
+            // This could be derived from ancestor branch + index part. Though the only caller of `timeline_init_and_sync` is `load_remote_timeline`,
+            // there will potentially be other caller of this function in the future, and we don't know whether `index_part` or `ancestor` takes precedence.
+            // Therefore, we pass this field explicitly for now, and remove it once we fully migrate to aux file v2.
+            last_aux_file_policy,
         )?;
         let disk_consistent_lsn = timeline.get_disk_consistent_lsn();
         anyhow::ensure!(
@@ -1176,12 +1182,15 @@ impl Tenant {
             None
         };
 
+        let last_aux_file_policy = index_part.last_aux_file_policy();
+
         self.timeline_init_and_sync(
             timeline_id,
             resources,
             Some(index_part),
             remote_metadata,
             ancestor,
+            last_aux_file_policy,
             ctx,
         )
         .await
@@ -1360,6 +1369,7 @@ impl Tenant {
             create_guard,
             initdb_lsn,
             None,
+            RuntimeAuxFilePolicy::Unspecified,
         )
         .await
     }
@@ -2431,6 +2441,7 @@ impl Tenant {
         ancestor: Option<Arc<Timeline>>,
         resources: TimelineResources,
         cause: CreateTimelineCause,
+        last_aux_file_policy: RuntimeAuxFilePolicy,
     ) -> anyhow::Result<Arc<Timeline>> {
         let state = match cause {
             CreateTimelineCause::Load => {
@@ -2459,6 +2470,7 @@ impl Tenant {
             resources,
             pg_version,
             state,
+            last_aux_file_policy,
             self.cancel.child_token(),
         );
 
@@ -3109,6 +3121,7 @@ impl Tenant {
                 timeline_create_guard,
                 start_lsn + 1,
                 Some(Arc::clone(src_timeline)),
+                src_timeline.last_aux_file_policy.load(),
             )
             .await?;
 
@@ -3311,6 +3324,7 @@ impl Tenant {
                 timeline_create_guard,
                 pgdata_lsn,
                 None,
+                RuntimeAuxFilePolicy::Unspecified,
             )
             .await?;
 
@@ -3388,6 +3402,7 @@ impl Tenant {
         create_guard: TimelineCreateGuard<'a>,
         start_lsn: Lsn,
         ancestor: Option<Arc<Timeline>>,
+        last_aux_file_policy: RuntimeAuxFilePolicy,
     ) -> anyhow::Result<UninitializedTimeline> {
         let tenant_shard_id = self.tenant_shard_id;
 
@@ -3403,6 +3418,7 @@ impl Tenant {
                 ancestor,
                 resources,
                 CreateTimelineCause::Load,
+                last_aux_file_policy,
             )
             .context("Failed to create timeline data structure")?;
 
diff --git a/pageserver/src/tenant/config.rs b/pageserver/src/tenant/config.rs
index a743ce3c16a3..b5a46713d98a 100644
--- a/pageserver/src/tenant/config.rs
+++ b/pageserver/src/tenant/config.rs
@@ -9,9 +9,9 @@
 //! may lead to a data loss.
 //!
 use anyhow::bail;
-use pageserver_api::models::AuxFilePolicy;
 use pageserver_api::models::CompactionAlgorithm;
 use pageserver_api::models::EvictionPolicy;
+use pageserver_api::models::SwitchAuxFilePolicy;
 use pageserver_api::models::{self, ThrottleConfig};
 use pageserver_api::shard::{ShardCount, ShardIdentity, ShardNumber, ShardStripeSize};
 use serde::de::IntoDeserializer;
@@ -373,7 +373,7 @@ pub struct TenantConf {
 
     /// 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.
-    pub switch_aux_file_policy: AuxFilePolicy,
+    pub switch_aux_file_policy: SwitchAuxFilePolicy,
 }
 
 /// Same as TenantConf, but this struct preserves the information about
@@ -472,7 +472,7 @@ pub struct TenantConfOpt {
 
     #[serde(skip_serializing_if = "Option::is_none")]
     #[serde(default)]
-    pub switch_aux_file_policy: Option<AuxFilePolicy>,
+    pub switch_aux_file_policy: Option<SwitchAuxFilePolicy>,
 }
 
 impl TenantConfOpt {
@@ -574,7 +574,7 @@ impl Default for TenantConf {
             lazy_slru_download: false,
             timeline_get_throttle: crate::tenant::throttle::Config::disabled(),
             image_layer_creation_check_threshold: DEFAULT_IMAGE_LAYER_CREATION_CHECK_THRESHOLD,
-            switch_aux_file_policy: AuxFilePolicy::V1,
+            switch_aux_file_policy: SwitchAuxFilePolicy::V1,
         }
     }
 }
diff --git a/pageserver/src/tenant/remote_timeline_client.rs b/pageserver/src/tenant/remote_timeline_client.rs
index 9103760388ce..97a1ee7f574f 100644
--- a/pageserver/src/tenant/remote_timeline_client.rs
+++ b/pageserver/src/tenant/remote_timeline_client.rs
@@ -189,6 +189,7 @@ use camino::Utf8Path;
 use chrono::{NaiveDateTime, Utc};
 
 pub(crate) use download::download_initdb_tar_zst;
+use pageserver_api::models::RuntimeAuxFilePolicy;
 use pageserver_api::shard::{ShardIndex, TenantShardId};
 use scopeguard::ScopeGuard;
 use tokio_util::sync::CancellationToken;
@@ -609,6 +610,17 @@ impl RemoteTimelineClient {
         Ok(())
     }
 
+    /// Launch an index-file upload operation in the background, with only aux_file_policy flag updated.
+    pub(crate) fn schedule_index_upload_for_aux_file_policy_update(
+        self: &Arc<Self>,
+        last_aux_file_policy: RuntimeAuxFilePolicy,
+    ) -> anyhow::Result<()> {
+        let mut guard = self.upload_queue.lock().unwrap();
+        let upload_queue = guard.initialized_mut()?;
+        upload_queue.last_aux_file_policy = last_aux_file_policy;
+        self.schedule_index_upload(upload_queue);
+        Ok(())
+    }
     ///
     /// Launch an index-file upload operation in the background, if necessary.
     ///
@@ -942,7 +954,7 @@ impl RemoteTimelineClient {
     }
 
     /// Wait for all previously scheduled uploads/deletions to complete
-    pub(crate) async fn wait_completion(self: &Arc<Self>) -> anyhow::Result<()> {
+    pub async fn wait_completion(self: &Arc<Self>) -> anyhow::Result<()> {
         let receiver = {
             let mut guard = self.upload_queue.lock().unwrap();
             let upload_queue = guard.initialized_mut()?;
@@ -1844,6 +1856,7 @@ impl RemoteTimelineClient {
                         dangling_files: HashMap::default(),
                         shutting_down: false,
                         shutdown_ready: Arc::new(tokio::sync::Semaphore::new(0)),
+                        last_aux_file_policy: initialized.last_aux_file_policy,
                     };
 
                     let upload_queue = std::mem::replace(
diff --git a/pageserver/src/tenant/remote_timeline_client/index.rs b/pageserver/src/tenant/remote_timeline_client/index.rs
index b114d6aa1047..b8f097ca88b7 100644
--- a/pageserver/src/tenant/remote_timeline_client/index.rs
+++ b/pageserver/src/tenant/remote_timeline_client/index.rs
@@ -5,6 +5,7 @@
 use std::collections::HashMap;
 
 use chrono::NaiveDateTime;
+use pageserver_api::models::RuntimeAuxFilePolicy;
 use serde::{Deserialize, Serialize};
 use utils::id::TimelineId;
 
@@ -88,6 +89,10 @@ pub struct IndexPart {
 
     #[serde(default)]
     pub(crate) lineage: Lineage,
+
+    // Added in version 6. None == RuntimeAuxFilePolicy::Unspecified. Use Option wrapper to keep forward compatibility.
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub last_aux_file_policy: Option<RuntimeAuxFilePolicy>,
 }
 
 impl IndexPart {
@@ -101,10 +106,11 @@ impl IndexPart {
     ///      is always generated from the keys of `layer_metadata`)
     /// - 4: timeline_layers is fully removed.
     /// - 5: lineage was added
-    const LATEST_VERSION: usize = 5;
+    /// - 6: last_aux_file_policy is added.
+    const LATEST_VERSION: usize = 6;
 
     // Versions we may see when reading from a bucket.
-    pub const KNOWN_VERSIONS: &'static [usize] = &[1, 2, 3, 4, 5];
+    pub const KNOWN_VERSIONS: &'static [usize] = &[1, 2, 3, 4, 5, 6];
 
     pub const FILE_NAME: &'static str = "index_part.json";
 
@@ -113,6 +119,7 @@ impl IndexPart {
         disk_consistent_lsn: Lsn,
         metadata: TimelineMetadata,
         lineage: Lineage,
+        last_aux_file_policy: RuntimeAuxFilePolicy,
     ) -> Self {
         let layer_metadata = layers_and_metadata
             .iter()
@@ -126,6 +133,7 @@ impl IndexPart {
             metadata,
             deleted_at: None,
             lineage,
+            last_aux_file_policy: Some(last_aux_file_policy),
         }
     }
 
@@ -155,8 +163,14 @@ impl IndexPart {
             example_metadata.disk_consistent_lsn(),
             example_metadata,
             Default::default(),
+            RuntimeAuxFilePolicy::V2,
         )
     }
+
+    pub fn last_aux_file_policy(&self) -> RuntimeAuxFilePolicy {
+        self.last_aux_file_policy
+            .unwrap_or(RuntimeAuxFilePolicy::Unspecified)
+    }
 }
 
 impl From<&UploadQueueInitialized> for IndexPart {
@@ -165,7 +179,13 @@ impl From<&UploadQueueInitialized> for IndexPart {
         let metadata = uq.latest_metadata.clone();
         let lineage = uq.latest_lineage.clone();
 
-        Self::new(&uq.latest_files, disk_consistent_lsn, metadata, lineage)
+        Self::new(
+            &uq.latest_files,
+            disk_consistent_lsn,
+            metadata,
+            lineage,
+            uq.last_aux_file_policy,
+        )
     }
 }
 
@@ -299,6 +319,7 @@ mod tests {
             metadata: TimelineMetadata::from_bytes(&[113,11,159,210,0,54,0,4,0,0,0,0,1,105,96,232,1,0,0,0,0,1,105,96,112,0,0,0,0,0,0,0,0,0,0,0,0,0,1,105,96,112,0,0,0,0,1,105,96,112,0,0,0,14,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]).unwrap(),
             deleted_at: None,
             lineage: Lineage::default(),
+            last_aux_file_policy: None,
         };
 
         let part = IndexPart::from_s3_bytes(example.as_bytes()).unwrap();
@@ -340,6 +361,7 @@ mod tests {
             metadata: TimelineMetadata::from_bytes(&[113,11,159,210,0,54,0,4,0,0,0,0,1,105,96,232,1,0,0,0,0,1,105,96,112,0,0,0,0,0,0,0,0,0,0,0,0,0,1,105,96,112,0,0,0,0,1,105,96,112,0,0,0,14,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]).unwrap(),
             deleted_at: None,
             lineage: Lineage::default(),
+            last_aux_file_policy: None,
         };
 
         let part = IndexPart::from_s3_bytes(example.as_bytes()).unwrap();
@@ -383,6 +405,7 @@ mod tests {
             deleted_at: Some(chrono::NaiveDateTime::parse_from_str(
                 "2023-07-31T09:00:00.123000000", "%Y-%m-%dT%H:%M:%S.%f").unwrap()),
             lineage: Lineage::default(),
+            last_aux_file_policy: None,
         };
 
         let part = IndexPart::from_s3_bytes(example.as_bytes()).unwrap();
@@ -428,6 +451,7 @@ mod tests {
             .unwrap(),
             deleted_at: None,
             lineage: Lineage::default(),
+            last_aux_file_policy: None,
         };
 
         let empty_layers_parsed = IndexPart::from_s3_bytes(empty_layers_json.as_bytes()).unwrap();
@@ -468,6 +492,7 @@ mod tests {
             metadata: TimelineMetadata::from_bytes(&[113,11,159,210,0,54,0,4,0,0,0,0,1,105,96,232,1,0,0,0,0,1,105,96,112,0,0,0,0,0,0,0,0,0,0,0,0,0,1,105,96,112,0,0,0,0,1,105,96,112,0,0,0,14,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]).unwrap(),
             deleted_at: Some(parse_naive_datetime("2023-07-31T09:00:00.123000000")),
             lineage: Lineage::default(),
+            last_aux_file_policy: None,
         };
 
         let part = IndexPart::from_s3_bytes(example.as_bytes()).unwrap();
@@ -511,6 +536,57 @@ mod tests {
                 reparenting_history: vec![TimelineId::from_str("e1bfd8c633d713d279e6fcd2bcc15b6d").unwrap()],
                 original_ancestor: Some((TimelineId::from_str("e2bfd8c633d713d279e6fcd2bcc15b6d").unwrap(), Lsn::from_str("0/15A7618").unwrap(), parse_naive_datetime("2024-05-07T18:52:36.322426563"))),
             },
+            last_aux_file_policy: None,
+        };
+
+        let part = IndexPart::from_s3_bytes(example.as_bytes()).unwrap();
+        assert_eq!(part, expected);
+    }
+
+    #[test]
+    fn v6_indexpart_is_parsed() {
+        let example = r#"{
+            "version":6,
+            "layer_metadata":{
+                "000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__0000000001696070-00000000016960E9": { "file_size": 25600000 },
+                "000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__00000000016B59D8-00000000016B5A51": { "file_size": 9007199254741001 }
+            },
+            "disk_consistent_lsn":"0/16960E8",
+            "metadata_bytes":[113,11,159,210,0,54,0,4,0,0,0,0,1,105,96,232,1,0,0,0,0,1,105,96,112,0,0,0,0,0,0,0,0,0,0,0,0,0,1,105,96,112,0,0,0,0,1,105,96,112,0,0,0,14,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0],
+            "deleted_at": "2023-07-31T09:00:00.123",
+            "lineage":{
+                "original_ancestor":["e2bfd8c633d713d279e6fcd2bcc15b6d","0/15A7618","2024-05-07T18:52:36.322426563"],
+                "reparenting_history":["e1bfd8c633d713d279e6fcd2bcc15b6d"]
+            },
+            "last_aux_file_policy": "V2"
+        }"#;
+
+        let expected = IndexPart {
+            version: 6,
+            layer_metadata: HashMap::from([
+                ("000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__0000000001696070-00000000016960E9".parse().unwrap(), IndexLayerMetadata {
+                    file_size: 25600000,
+                    generation: Generation::none(),
+                    shard: ShardIndex::unsharded()
+                }),
+                ("000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__00000000016B59D8-00000000016B5A51".parse().unwrap(), IndexLayerMetadata {
+                    // serde_json should always parse this but this might be a double with jq for
+                    // example.
+                    file_size: 9007199254741001,
+                    generation: Generation::none(),
+                    shard: ShardIndex::unsharded()
+                })
+            ]),
+            disk_consistent_lsn: "0/16960E8".parse::<Lsn>().unwrap(),
+            metadata: TimelineMetadata::from_bytes(&[113,11,159,210,0,54,0,4,0,0,0,0,1,105,96,232,1,0,0,0,0,1,105,96,112,0,0,0,0,0,0,0,0,0,0,0,0,0,1,105,96,112,0,0,0,0,1,105,96,112,0,0,0,14,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]).unwrap(),
+            deleted_at: Some(chrono::NaiveDateTime::parse_from_str(
+                "2023-07-31T09:00:00.123000000", "%Y-%m-%dT%H:%M:%S.%f").unwrap()),
+            lineage: Lineage {
+                reparenting_history_truncated: false,
+                reparenting_history: vec![TimelineId::from_str("e1bfd8c633d713d279e6fcd2bcc15b6d").unwrap()],
+                original_ancestor: Some((TimelineId::from_str("e2bfd8c633d713d279e6fcd2bcc15b6d").unwrap(), Lsn::from_str("0/15A7618").unwrap(), parse_naive_datetime("2024-05-07T18:52:36.322426563"))),
+            },
+            last_aux_file_policy: Some(RuntimeAuxFilePolicy::V2),
         };
 
         let part = IndexPart::from_s3_bytes(example.as_bytes()).unwrap();
diff --git a/pageserver/src/tenant/timeline.rs b/pageserver/src/tenant/timeline.rs
index 01f354b9e85d..f80546dfd3cc 100644
--- a/pageserver/src/tenant/timeline.rs
+++ b/pageserver/src/tenant/timeline.rs
@@ -23,9 +23,9 @@ use pageserver_api::{
     },
     keyspace::{KeySpaceAccum, SparseKeyPartitioning},
     models::{
-        AuxFilePolicy, CompactionAlgorithm, DownloadRemoteLayersTaskInfo,
+        AtomicRuntimeAuxFilePolicy, CompactionAlgorithm, DownloadRemoteLayersTaskInfo,
         DownloadRemoteLayersTaskSpawnRequest, EvictionPolicy, InMemoryLayerInfo, LayerMapInfo,
-        TimelineState,
+        RuntimeAuxFilePolicy, SwitchAuxFilePolicy, TimelineState,
     },
     reltag::BlockNumber,
     shard::{ShardIdentity, ShardNumber, TenantShardId},
@@ -413,7 +413,11 @@ pub struct Timeline {
     /// Keep aux directory cache to avoid it's reconstruction on each update
     pub(crate) aux_files: tokio::sync::Mutex<AuxFilesState>,
 
+    /// Size estiamtor for aux file v2
     pub(crate) aux_file_size_estimator: AuxFileSizeEstimator,
+
+    /// Indicate whether aux file v2 storage is enabled.
+    pub(crate) last_aux_file_policy: AtomicRuntimeAuxFilePolicy,
 }
 
 pub struct WalReceiverInfo {
@@ -2011,7 +2015,7 @@ const REPARTITION_FREQ_IN_CHECKPOINT_DISTANCE: u64 = 10;
 
 // Private functions
 impl Timeline {
-    pub(crate) fn get_switch_aux_file_policy(&self) -> AuxFilePolicy {
+    pub(crate) fn get_switch_aux_file_policy(&self) -> SwitchAuxFilePolicy {
         let tenant_conf = self.tenant_conf.load();
         tenant_conf
             .tenant_conf
@@ -2149,6 +2153,7 @@ impl Timeline {
         resources: TimelineResources,
         pg_version: u32,
         state: TimelineState,
+        aux_file_policy: RuntimeAuxFilePolicy,
         cancel: CancellationToken,
     ) -> Arc<Self> {
         let disk_consistent_lsn = metadata.disk_consistent_lsn();
@@ -2273,6 +2278,8 @@ impl Timeline {
                 }),
 
                 aux_file_size_estimator: AuxFileSizeEstimator::new(aux_file_metrics),
+
+                last_aux_file_policy: AtomicRuntimeAuxFilePolicy::new(aux_file_policy),
             };
             result.repartition_threshold =
                 result.get_checkpoint_distance() / REPARTITION_FREQ_IN_CHECKPOINT_DISTANCE;
@@ -2422,6 +2429,11 @@ impl Timeline {
         let shard = self.get_shard_index();
         let this = self.myself.upgrade().expect("&self method holds the arc");
 
+        if let Some(ref index_part) = index_part {
+            self.last_aux_file_policy
+                .store(index_part.last_aux_file_policy());
+        }
+
         let (loaded_layers, needs_cleanup, total_physical_size) = tokio::task::spawn_blocking({
             move || {
                 let _g = span.entered();
diff --git a/pageserver/src/tenant/timeline/delete.rs b/pageserver/src/tenant/timeline/delete.rs
index d8701be17013..471a1e50310b 100644
--- a/pageserver/src/tenant/timeline/delete.rs
+++ b/pageserver/src/tenant/timeline/delete.rs
@@ -4,7 +4,10 @@ use std::{
 };
 
 use anyhow::Context;
-use pageserver_api::{models::TimelineState, shard::TenantShardId};
+use pageserver_api::{
+    models::{RuntimeAuxFilePolicy, TimelineState},
+    shard::TenantShardId,
+};
 use tokio::sync::OwnedMutexGuard;
 use tracing::{error, info, instrument, Instrument};
 use utils::{crashsafe, fs_ext, id::TimelineId};
@@ -278,6 +281,7 @@ impl DeleteTimelineFlow {
                 // Important. We dont pass ancestor above because it can be missing.
                 // Thus we need to skip the validation here.
                 CreateTimelineCause::Delete,
+                RuntimeAuxFilePolicy::Unspecified, // Aux file policy is not needed for deletion, assuming deletion does not read aux keyspace
             )
             .context("create_timeline_struct")?;
 
diff --git a/pageserver/src/tenant/upload_queue.rs b/pageserver/src/tenant/upload_queue.rs
index a2f761fa949a..f4f9c286f129 100644
--- a/pageserver/src/tenant/upload_queue.rs
+++ b/pageserver/src/tenant/upload_queue.rs
@@ -8,6 +8,7 @@ use std::collections::{HashMap, VecDeque};
 use std::fmt::Debug;
 
 use chrono::NaiveDateTime;
+use pageserver_api::models::RuntimeAuxFilePolicy;
 use std::sync::Arc;
 use tracing::info;
 use utils::lsn::AtomicLsn;
@@ -60,6 +61,8 @@ pub(crate) struct UploadQueueInitialized {
     /// Part of the flattened "next" `index_part.json`.
     pub(crate) latest_lineage: Lineage,
 
+    pub(crate) last_aux_file_policy: RuntimeAuxFilePolicy,
+
     /// `disk_consistent_lsn` from the last metadata file that was successfully
     /// uploaded. `Lsn(0)` if nothing was uploaded yet.
     /// Unlike `latest_files` or `latest_metadata`, this value is never ahead.
@@ -189,6 +192,7 @@ impl UploadQueue {
             dangling_files: HashMap::new(),
             shutting_down: false,
             shutdown_ready: Arc::new(tokio::sync::Semaphore::new(0)),
+            last_aux_file_policy: RuntimeAuxFilePolicy::Unspecified,
         };
 
         *self = UploadQueue::Initialized(state);
@@ -239,6 +243,7 @@ impl UploadQueue {
             dangling_files: HashMap::new(),
             shutting_down: false,
             shutdown_ready: Arc::new(tokio::sync::Semaphore::new(0)),
+            last_aux_file_policy: index_part.last_aux_file_policy(),
         };
 
         *self = UploadQueue::Initialized(state);
diff --git a/test_runner/regress/test_aux_files.py b/test_runner/regress/test_aux_files.py
new file mode 100644
index 000000000000..139d49c9f23d
--- /dev/null
+++ b/test_runner/regress/test_aux_files.py
@@ -0,0 +1,81 @@
+from fixtures.log_helper import log
+from fixtures.neon_fixtures import (
+    NeonEnv,
+    logical_replication_sync,
+)
+
+
+def test_aux_v2_config_switch(neon_simple_env: NeonEnv, vanilla_pg):
+    env = neon_simple_env
+
+    tenant_id = env.initial_tenant
+    timeline_id = env.neon_cli.create_branch("test_aux_v2_config_switch", "empty")
+    endpoint = env.endpoints.create_start(
+        "test_aux_v2_config_switch", config_lines=["log_statement=all"]
+    )
+
+    with env.pageserver.http_client() as client:
+        tenant_config = client.tenant_config(tenant_id).effective_config
+        tenant_config["switch_aux_file_policy"] = "V2"
+        client.set_tenant_config(tenant_id, tenant_config)
+        # aux file v2 is enabled on the write path
+        assert (
+            client.timeline_detail(tenant_id=tenant_id, timeline_id=timeline_id)[
+                "last_aux_file_policy"
+            ]
+            == "Unspecified"
+        )
+    pg_conn = endpoint.connect()
+    cur = pg_conn.cursor()
+
+    cur.execute("create table t(pk integer primary key, payload integer)")
+    cur.execute(
+        "CREATE TABLE replication_example(id SERIAL PRIMARY KEY, somedata int, text varchar(120));"
+    )
+    cur.execute("create publication pub1 for table t, replication_example")
+
+    # now start subscriber, aux files will be created at this point. TODO: find better ways of testing aux files (i.e., neon_test_utils)
+    # instead of going through the full logical replication process.
+    vanilla_pg.start()
+    vanilla_pg.safe_psql("create table t(pk integer primary key, payload integer)")
+    vanilla_pg.safe_psql(
+        "CREATE TABLE replication_example(id SERIAL PRIMARY KEY, somedata int, text varchar(120), testcolumn1 int, testcolumn2 int, testcolumn3 int);"
+    )
+    connstr = endpoint.connstr().replace("'", "''")
+    log.info(f"ep connstr is {endpoint.connstr()}, subscriber connstr {vanilla_pg.connstr()}")
+    vanilla_pg.safe_psql(f"create subscription sub1 connection '{connstr}' publication pub1")
+
+    # Wait logical replication channel to be established
+    logical_replication_sync(vanilla_pg, endpoint)
+    vanilla_pg.stop()
+    endpoint.stop()
+
+    with env.pageserver.http_client() as client:
+        # aux file v2 flag should be enabled at this point
+        assert (
+            client.timeline_detail(tenant_id=tenant_id, timeline_id=timeline_id)[
+                "last_aux_file_policy"
+            ]
+            == "V2"
+        )
+    with env.pageserver.http_client() as client:
+        tenant_config = client.tenant_config(tenant_id).effective_config
+        tenant_config["switch_aux_file_policy"] = "V1"
+        client.set_tenant_config(tenant_id, tenant_config)
+        # the flag should still be enabled
+        assert (
+            client.timeline_detail(tenant_id=tenant_id, timeline_id=timeline_id)[
+                "last_aux_file_policy"
+            ]
+            == "V2"
+        )
+    env.pageserver.restart()
+    with env.pageserver.http_client() as client:
+        # aux file v2 flag should be persisted
+        assert (
+            client.timeline_detail(tenant_id=tenant_id, timeline_id=timeline_id)[
+                "last_aux_file_policy"
+            ]
+            == "V2"
+        )
+    # TODO(chi): test with timeline detach?

From 4c82a9dda97ef26ef0a451eac543c74905aa501f Mon Sep 17 00:00:00 2001
From: Alex Chi Z <chi@neon.tech>
Date: Tue, 14 May 2024 10:41:52 -0400
Subject: [PATCH 02/20] resolve comments

Signed-off-by: Alex Chi Z <chi@neon.tech>
---
 libs/pageserver_api/src/models.rs        | 29 +++++++++++++++---------
 pageserver/src/tenant/timeline/delete.rs |  3 ++-
 2 files changed, 20 insertions(+), 12 deletions(-)

diff --git a/libs/pageserver_api/src/models.rs b/libs/pageserver_api/src/models.rs
index 6771bd9e19a8..22eababf3cb6 100644
--- a/libs/pageserver_api/src/models.rs
+++ b/libs/pageserver_api/src/models.rs
@@ -352,14 +352,17 @@ impl SwitchAuxFilePolicy {
         }
     }
 
-    pub fn from_usize(this: usize) -> Self {
+    pub fn try_from_usize(this: usize) -> Option<Self> {
         match this {
-            1 => Self::V1,
-            2 => Self::CrossValidation,
-            3 => Self::V2,
-            _ => unreachable!(),
+            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()
+    }
 
     pub fn to_runtime_policy(&self) -> RuntimeAuxFilePolicy {
         RuntimeAuxFilePolicy::from_usize(self.to_usize())
@@ -376,15 +379,19 @@ impl RuntimeAuxFilePolicy {
         }
     }
 
-    pub fn from_usize(this: usize) -> Self {
+    pub fn try_from_usize(this: usize) -> Option<Self> {
         match this {
-            1 => Self::V1,
-            2 => Self::CrossValidation,
-            3 => Self::V2,
-            0 => Self::Unspecified,
-            _ => unreachable!(),
+            1 => Some(Self::V1),
+            2 => Some(Self::CrossValidation),
+            3 => Some(Self::V2),
+            0 => Some(Self::Unspecified),
+            _ => None,
         }
     }
+
+    pub fn from_usize(this: usize) -> Self {
+        Self::try_from_usize(this).unwrap()
+    }
 }
 
 impl FromStr for RuntimeAuxFilePolicy {
diff --git a/pageserver/src/tenant/timeline/delete.rs b/pageserver/src/tenant/timeline/delete.rs
index 471a1e50310b..dcbc9aadf03f 100644
--- a/pageserver/src/tenant/timeline/delete.rs
+++ b/pageserver/src/tenant/timeline/delete.rs
@@ -281,7 +281,8 @@ impl DeleteTimelineFlow {
                 // Important. We dont pass ancestor above because it can be missing.
                 // Thus we need to skip the validation here.
                 CreateTimelineCause::Delete,
-                RuntimeAuxFilePolicy::Unspecified, // Aux file policy is not needed for deletion, assuming deletion does not read aux keyspace
+                // Aux file policy is not needed for deletion, assuming deletion does not read aux keyspace
+                RuntimeAuxFilePolicy::Unspecified,
             )
             .context("create_timeline_struct")?;
 

From 54dd764909a645c6c25cab1bebd949db771e7d3a Mon Sep 17 00:00:00 2001
From: Alex Chi Z <chi@neon.tech>
Date: Tue, 14 May 2024 15:25:47 -0400
Subject: [PATCH 03/20] resolve comments

Signed-off-by: Alex Chi Z <chi@neon.tech>
---
 libs/pageserver_api/src/models.rs                     | 2 +-
 pageserver/src/pgdatadir_mapping.rs                   | 2 +-
 pageserver/src/tenant/config.rs                       | 2 ++
 pageserver/src/tenant/remote_timeline_client.rs       | 2 +-
 pageserver/src/tenant/remote_timeline_client/index.rs | 8 +++++---
 pageserver/src/tenant/timeline.rs                     | 2 +-
 6 files changed, 11 insertions(+), 7 deletions(-)

diff --git a/libs/pageserver_api/src/models.rs b/libs/pageserver_api/src/models.rs
index 22eababf3cb6..18858f098524 100644
--- a/libs/pageserver_api/src/models.rs
+++ b/libs/pageserver_api/src/models.rs
@@ -704,7 +704,7 @@ pub struct TimelineInfo {
     pub walreceiver_status: String,
 
     /// Whether aux file v2 is enabled
-    pub last_aux_file_policy: RuntimeAuxFilePolicy,
+    pub(crate) last_aux_file_policy: RuntimeAuxFilePolicy,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
diff --git a/pageserver/src/pgdatadir_mapping.rs b/pageserver/src/pgdatadir_mapping.rs
index 3bd06eee35dd..e5f36fdea629 100644
--- a/pageserver/src/pgdatadir_mapping.rs
+++ b/pageserver/src/pgdatadir_mapping.rs
@@ -1497,7 +1497,7 @@ impl<'a> DatadirModification<'a> {
             }
         };
 
-        assert!(policy != RuntimeAuxFilePolicy::Unspecified);
+        assert_ne!(policy, RuntimeAuxFilePolicy::Unspecified);
 
         if let RuntimeAuxFilePolicy::V2 | RuntimeAuxFilePolicy::CrossValidation = policy {
             let key = aux_file::encode_aux_file_key(path);
diff --git a/pageserver/src/tenant/config.rs b/pageserver/src/tenant/config.rs
index b5a46713d98a..c328ee8e63b4 100644
--- a/pageserver/src/tenant/config.rs
+++ b/pageserver/src/tenant/config.rs
@@ -373,6 +373,8 @@ pub struct TenantConf {
 
     /// 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: SwitchAuxFilePolicy,
 }
 
diff --git a/pageserver/src/tenant/remote_timeline_client.rs b/pageserver/src/tenant/remote_timeline_client.rs
index 97a1ee7f574f..2ab59f90c3d2 100644
--- a/pageserver/src/tenant/remote_timeline_client.rs
+++ b/pageserver/src/tenant/remote_timeline_client.rs
@@ -954,7 +954,7 @@ impl RemoteTimelineClient {
     }
 
     /// Wait for all previously scheduled uploads/deletions to complete
-    pub async fn wait_completion(self: &Arc<Self>) -> anyhow::Result<()> {
+    pub(crate) async fn wait_completion(self: &Arc<Self>) -> anyhow::Result<()> {
         let receiver = {
             let mut guard = self.upload_queue.lock().unwrap();
             let upload_queue = guard.initialized_mut()?;
diff --git a/pageserver/src/tenant/remote_timeline_client/index.rs b/pageserver/src/tenant/remote_timeline_client/index.rs
index b8f097ca88b7..bd916b1a84db 100644
--- a/pageserver/src/tenant/remote_timeline_client/index.rs
+++ b/pageserver/src/tenant/remote_timeline_client/index.rs
@@ -91,8 +91,10 @@ pub struct IndexPart {
     pub(crate) lineage: Lineage,
 
     // Added in version 6. None == RuntimeAuxFilePolicy::Unspecified. Use Option wrapper to keep forward compatibility.
+    // This flag is controlled by TenantConf::switch_aux_file_policy. When the first aux file gets written, the aux file
+    // mode will be persisted in `index_part.json`, and `switch_aux_file_policy` will be ignored.
     #[serde(skip_serializing_if = "Option::is_none", default)]
-    pub last_aux_file_policy: Option<RuntimeAuxFilePolicy>,
+    pub(crate) last_aux_file_policy: Option<RuntimeAuxFilePolicy>,
 }
 
 impl IndexPart {
@@ -163,11 +165,11 @@ impl IndexPart {
             example_metadata.disk_consistent_lsn(),
             example_metadata,
             Default::default(),
-            RuntimeAuxFilePolicy::V2,
+            RuntimeAuxFilePolicy::V1,
         )
     }
 
-    pub fn last_aux_file_policy(&self) -> RuntimeAuxFilePolicy {
+    pub(crate) fn last_aux_file_policy(&self) -> RuntimeAuxFilePolicy {
         self.last_aux_file_policy
             .unwrap_or(RuntimeAuxFilePolicy::Unspecified)
     }
diff --git a/pageserver/src/tenant/timeline.rs b/pageserver/src/tenant/timeline.rs
index f80546dfd3cc..b54819434b9b 100644
--- a/pageserver/src/tenant/timeline.rs
+++ b/pageserver/src/tenant/timeline.rs
@@ -413,7 +413,7 @@ pub struct Timeline {
     /// Keep aux directory cache to avoid it's reconstruction on each update
     pub(crate) aux_files: tokio::sync::Mutex<AuxFilesState>,
 
-    /// Size estiamtor for aux file v2
+    /// Size estimator for aux file v2
     pub(crate) aux_file_size_estimator: AuxFileSizeEstimator,
 
     /// Indicate whether aux file v2 storage is enabled.

From 1d36ec757a857c7deb6612e84f048a4a20c53db1 Mon Sep 17 00:00:00 2001
From: Alex Chi Z <chi@neon.tech>
Date: Wed, 15 May 2024 09:27:53 -0400
Subject: [PATCH 04/20] fix visibility

Signed-off-by: Alex Chi Z <chi@neon.tech>
---
 pageserver/src/tenant/timeline.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/pageserver/src/tenant/timeline.rs b/pageserver/src/tenant/timeline.rs
index b54819434b9b..75290f66078c 100644
--- a/pageserver/src/tenant/timeline.rs
+++ b/pageserver/src/tenant/timeline.rs
@@ -417,7 +417,7 @@ pub struct Timeline {
     pub(crate) aux_file_size_estimator: AuxFileSizeEstimator,
 
     /// Indicate whether aux file v2 storage is enabled.
-    pub(crate) last_aux_file_policy: AtomicRuntimeAuxFilePolicy,
+    pub last_aux_file_policy: AtomicRuntimeAuxFilePolicy,
 }
 
 pub struct WalReceiverInfo {

From d0bfcd2ef7572cf4095f0afc22413d76b483e9cc Mon Sep 17 00:00:00 2001
From: Alex Chi Z <chi@neon.tech>
Date: Wed, 15 May 2024 14:11:26 -0400
Subject: [PATCH 05/20] rm RuntimeAuxFilePolicy

Signed-off-by: Alex Chi Z <chi@neon.tech>
---
 control_plane/src/pageserver.rs               |   6 +-
 libs/pageserver_api/src/models.rs             | 104 +++++++-----------
 pageserver/src/pgdatadir_mapping.rs           |  28 ++---
 pageserver/src/tenant.rs                      |  12 +-
 pageserver/src/tenant/config.rs               |   8 +-
 .../src/tenant/remote_timeline_client.rs      |   4 +-
 .../tenant/remote_timeline_client/index.rs    |  17 ++-
 pageserver/src/tenant/timeline.rs             |  12 +-
 pageserver/src/tenant/timeline/delete.rs      |   7 +-
 pageserver/src/tenant/upload_queue.rs         |   7 +-
 test_runner/regress/test_aux_files.py         |   5 +-
 11 files changed, 85 insertions(+), 125 deletions(-)

diff --git a/control_plane/src/pageserver.rs b/control_plane/src/pageserver.rs
index a8fa8486e8c5..4b2b0348290d 100644
--- a/control_plane/src/pageserver.rs
+++ b/control_plane/src/pageserver.rs
@@ -17,7 +17,7 @@ use anyhow::{bail, Context};
 use camino::Utf8PathBuf;
 use futures::SinkExt;
 use pageserver_api::models::{
-    self, LocationConfig, ShardParameters, SwitchAuxFilePolicy, TenantHistorySize, TenantInfo,
+    self, LocationConfig, ShardParameters, AuxFilePolicy, TenantHistorySize, TenantInfo,
     TimelineInfo,
 };
 use pageserver_api::shard::TenantShardId;
@@ -380,7 +380,7 @@ impl PageServerNode {
                 .context("parse `timeline_get_throttle` from json")?,
             switch_aux_file_policy: settings
                 .remove("switch_aux_file_policy")
-                .map(|x| x.parse::<SwitchAuxFilePolicy>())
+                .map(|x| x.parse::<AuxFilePolicy>())
                 .transpose()
                 .context("Failed to parse 'switch_aux_file_policy'")?,
         };
@@ -503,7 +503,7 @@ impl PageServerNode {
                     .context("parse `timeline_get_throttle` from json")?,
                 switch_aux_file_policy: settings
                     .remove("switch_aux_file_policy")
-                    .map(|x| x.parse::<SwitchAuxFilePolicy>())
+                    .map(|x| x.parse::<AuxFilePolicy>())
                     .transpose()
                     .context("Failed to parse 'switch_aux_file_policy'")?,
             }
diff --git a/libs/pageserver_api/src/models.rs b/libs/pageserver_api/src/models.rs
index 18858f098524..e61215cb692e 100644
--- a/libs/pageserver_api/src/models.rs
+++ b/libs/pageserver_api/src/models.rs
@@ -306,44 +306,57 @@ 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<SwitchAuxFilePolicy>,
-}
-
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
-pub enum SwitchAuxFilePolicy {
-    V1,
-    V2,
-    CrossValidation,
+    pub switch_aux_file_policy: Option<AuxFilePolicy>,
 }
 
+/// The policy for the aux file storage. It can be switched through `switch_aux_file_policy`
+/// tenant config. When the first aux file written, the policy will be persisted in the
+/// `index_part.json` file and has a limited migration path.
+///
+/// Currently, we only allow the following migration path:
+///
+/// Unset -> V1
+///       -> V2
+///       -> CrossValidation -> V2
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
-pub enum RuntimeAuxFilePolicy {
+pub enum AuxFilePolicy {
+    /// V1 aux file policy: store everything in AUX_FILE_KEY
     V1,
+    /// V2 aux file policy: store in the AUX_FILE keyspace
     V2,
+    /// Cross validation runs both formats on the write path and does validation
+    /// on the read path.
     CrossValidation,
-    Unspecified,
 }
 
-pub struct AtomicRuntimeAuxFilePolicy(AtomicUsize);
+/// The aux file policy memory flag. Users can store `Option<AuxFilePolicy>` into this atomic flag. 0 == unspecified.
+pub struct AtomicAuxFilePolicy(AtomicUsize);
 
-impl AtomicRuntimeAuxFilePolicy {
-    pub fn new(policy: RuntimeAuxFilePolicy) -> Self {
-        Self(AtomicUsize::new(policy.to_usize()))
+impl AtomicAuxFilePolicy {
+    pub fn new(policy: Option<AuxFilePolicy>) -> Self {
+        Self(AtomicUsize::new(
+            policy.map(AuxFilePolicy::to_usize).unwrap_or_default(),
+        ))
     }
 
-    pub fn load(&self) -> RuntimeAuxFilePolicy {
-        RuntimeAuxFilePolicy::from_usize(self.0.load(std::sync::atomic::Ordering::SeqCst))
+    pub fn load(&self) -> Option<AuxFilePolicy> {
+        // Using SeqCst memory order because this flag will be read from both write/read threads
+        // and we want to guarantee the strongest consistency.
+        match self.0.load(std::sync::atomic::Ordering::SeqCst) {
+            0 => None,
+            other => Some(AuxFilePolicy::from_usize(other)),
+        }
     }
 
-    pub fn store(&self, policy: RuntimeAuxFilePolicy) {
+    pub fn store(&self, policy: Option<AuxFilePolicy>) {
         self.0.store(
-            RuntimeAuxFilePolicy::to_usize(policy),
+            policy.map(AuxFilePolicy::to_usize).unwrap_or_default(),
             std::sync::atomic::Ordering::SeqCst,
         );
     }
 }
 
-impl SwitchAuxFilePolicy {
+impl AuxFilePolicy {
     pub fn to_usize(self) -> usize {
         match self {
             Self::V1 => 1,
@@ -360,60 +373,17 @@ impl SwitchAuxFilePolicy {
             _ => None,
         }
     }
-    pub fn from_usize(this: usize) -> Self {
-        Self::try_from_usize(this).unwrap()
-    }
-
-    pub fn to_runtime_policy(&self) -> RuntimeAuxFilePolicy {
-        RuntimeAuxFilePolicy::from_usize(self.to_usize())
-    }
-}
-
-impl RuntimeAuxFilePolicy {
-    pub fn to_usize(self) -> usize {
-        match self {
-            Self::V1 => 1,
-            Self::CrossValidation => 2,
-            Self::V2 => 3,
-            Self::Unspecified => 0,
-        }
-    }
-
-    pub fn try_from_usize(this: usize) -> Option<Self> {
-        match this {
-            1 => Some(Self::V1),
-            2 => Some(Self::CrossValidation),
-            3 => Some(Self::V2),
-            0 => Some(Self::Unspecified),
-            _ => None,
-        }
-    }
 
     pub fn from_usize(this: usize) -> Self {
         Self::try_from_usize(this).unwrap()
     }
-}
-
-impl FromStr for RuntimeAuxFilePolicy {
-    type Err = anyhow::Error;
 
-    fn from_str(s: &str) -> Result<Self, Self::Err> {
-        let s = s.to_lowercase();
-        if s.is_empty() || s == "unspecified" {
-            Ok(Self::Unspecified)
-        } else if s == "v1" {
-            Ok(Self::V1)
-        } else if s == "v2" {
-            Ok(Self::V2)
-        } else if s == "crossvalidation" || s == "cross_validation" {
-            Ok(Self::CrossValidation)
-        } else {
-            anyhow::bail!("cannot parse {} to aux file policy", s)
-        }
+    pub fn to_runtime_policy(&self) -> AuxFilePolicy {
+        AuxFilePolicy::from_usize(self.to_usize())
     }
 }
 
-impl FromStr for SwitchAuxFilePolicy {
+impl FromStr for AuxFilePolicy {
     type Err = anyhow::Error;
 
     fn from_str(s: &str) -> Result<Self, Self::Err> {
@@ -703,8 +673,8 @@ pub struct TimelineInfo {
 
     pub walreceiver_status: String,
 
-    /// Whether aux file v2 is enabled
-    pub(crate) last_aux_file_policy: RuntimeAuxFilePolicy,
+    /// The last aux file policy being used on this timeline
+    pub last_aux_file_policy: Option<AuxFilePolicy>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
diff --git a/pageserver/src/pgdatadir_mapping.rs b/pageserver/src/pgdatadir_mapping.rs
index e5f36fdea629..c7da9211baa6 100644
--- a/pageserver/src/pgdatadir_mapping.rs
+++ b/pageserver/src/pgdatadir_mapping.rs
@@ -24,7 +24,7 @@ use pageserver_api::key::{
     AUX_FILES_KEY, CHECKPOINT_KEY, CONTROLFILE_KEY, DBDIR_KEY, TWOPHASEDIR_KEY,
 };
 use pageserver_api::keyspace::SparseKeySpace;
-use pageserver_api::models::{RuntimeAuxFilePolicy, SwitchAuxFilePolicy};
+use pageserver_api::models::{AuxFilePolicy, };
 use pageserver_api::reltag::{BlockNumber, RelTag, SlruKind};
 use postgres_ffi::relfile_utils::{FSM_FORKNUM, VISIBILITYMAP_FORKNUM};
 use postgres_ffi::BLCKSZ;
@@ -720,11 +720,11 @@ impl Timeline {
     ) -> Result<HashMap<String, Bytes>, PageReconstructError> {
         let current_policy = self.last_aux_file_policy.load();
         match current_policy {
-            RuntimeAuxFilePolicy::V1 | RuntimeAuxFilePolicy::Unspecified => {
+            Some(AuxFilePolicy::V1) | None => {
                 self.list_aux_files_v1(lsn, ctx).await
             }
-            RuntimeAuxFilePolicy::V2 => self.list_aux_files_v2(lsn, ctx).await,
-            RuntimeAuxFilePolicy::CrossValidation => {
+            Some(AuxFilePolicy::V2) => self.list_aux_files_v2(lsn, ctx).await,
+            Some(AuxFilePolicy::CrossValidation) => {
                 let v1_result = self.list_aux_files_v1(lsn, ctx).await;
                 let v2_result = self.list_aux_files_v2(lsn, ctx).await;
                 match (v1_result, v2_result) {
@@ -1454,7 +1454,7 @@ impl<'a> DatadirModification<'a> {
     }
 
     pub fn init_aux_dir(&mut self) -> anyhow::Result<()> {
-        if let SwitchAuxFilePolicy::V2 = self.tline.get_switch_aux_file_policy() {
+        if let AuxFilePolicy::V2 = self.tline.get_switch_aux_file_policy() {
             return Ok(());
         }
         let buf = AuxFilesDirectory::ser(&AuxFilesDirectory {
@@ -1480,26 +1480,22 @@ impl<'a> DatadirModification<'a> {
             // * no aux files -> v1/v2/cross-validation
             // * cross-validation->v2
             match (current_policy, switch_policy) {
-                (RuntimeAuxFilePolicy::Unspecified, _)
-                | (RuntimeAuxFilePolicy::CrossValidation, SwitchAuxFilePolicy::V2) => {
+                (None, _)
+                | (Some(AuxFilePolicy::CrossValidation), AuxFilePolicy::V2) => {
                     let new_policy = switch_policy.to_runtime_policy();
-                    self.tline.last_aux_file_policy.store(new_policy);
+                    self.tline.last_aux_file_policy.store(Some(new_policy));
                     info!("switching to aux file policy {:?}", new_policy);
                     if let Some(ref remote_client) = self.tline.remote_client {
                         remote_client
-                            .schedule_index_upload_for_aux_file_policy_update(new_policy)?;
-                        remote_client.wait_completion().await?;
+                            .schedule_index_upload_for_aux_file_policy_update(Some(new_policy))?;
                     }
-                    info!("finish switching to aux file policy {:?}", new_policy);
                     new_policy
                 }
-                (current_policy, _) => current_policy,
+                (Some(current_policy), _) => current_policy,
             }
         };
 
-        assert_ne!(policy, RuntimeAuxFilePolicy::Unspecified);
-
-        if let RuntimeAuxFilePolicy::V2 | RuntimeAuxFilePolicy::CrossValidation = policy {
+        if let AuxFilePolicy::V2 | AuxFilePolicy::CrossValidation = policy {
             let key = aux_file::encode_aux_file_key(path);
             // retrieve the key from the engine
             let old_val = match self.get(key, ctx).await {
@@ -1550,7 +1546,7 @@ impl<'a> DatadirModification<'a> {
             self.put(key, Value::Image(new_val.into()));
         }
 
-        if let RuntimeAuxFilePolicy::V1 | RuntimeAuxFilePolicy::CrossValidation = policy {
+        if let AuxFilePolicy::V1 | AuxFilePolicy::CrossValidation = policy {
             let file_path = path.to_string();
             let content = if content.is_empty() {
                 None
diff --git a/pageserver/src/tenant.rs b/pageserver/src/tenant.rs
index bfe8ea57688b..b605961b4590 100644
--- a/pageserver/src/tenant.rs
+++ b/pageserver/src/tenant.rs
@@ -20,7 +20,7 @@ use futures::stream::FuturesUnordered;
 use futures::FutureExt;
 use futures::StreamExt;
 use pageserver_api::models;
-use pageserver_api::models::RuntimeAuxFilePolicy;
+use pageserver_api::models::AuxFilePolicy;
 use pageserver_api::models::TimelineState;
 use pageserver_api::models::WalRedoManagerStatus;
 use pageserver_api::shard::ShardIdentity;
@@ -529,7 +529,7 @@ impl Tenant {
         index_part: Option<IndexPart>,
         metadata: TimelineMetadata,
         ancestor: Option<Arc<Timeline>>,
-        last_aux_file_policy: RuntimeAuxFilePolicy,
+        last_aux_file_policy: Option<AuxFilePolicy>,
         _ctx: &RequestContext,
     ) -> anyhow::Result<()> {
         let tenant_id = self.tenant_shard_id;
@@ -1369,7 +1369,7 @@ impl Tenant {
             create_guard,
             initdb_lsn,
             None,
-            RuntimeAuxFilePolicy::Unspecified,
+            None,
         )
         .await
     }
@@ -2441,7 +2441,7 @@ impl Tenant {
         ancestor: Option<Arc<Timeline>>,
         resources: TimelineResources,
         cause: CreateTimelineCause,
-        last_aux_file_policy: RuntimeAuxFilePolicy,
+        last_aux_file_policy: Option<AuxFilePolicy>,
     ) -> anyhow::Result<Arc<Timeline>> {
         let state = match cause {
             CreateTimelineCause::Load => {
@@ -3324,7 +3324,7 @@ impl Tenant {
                 timeline_create_guard,
                 pgdata_lsn,
                 None,
-                RuntimeAuxFilePolicy::Unspecified,
+                None,
             )
             .await?;
 
@@ -3402,7 +3402,7 @@ impl Tenant {
         create_guard: TimelineCreateGuard<'a>,
         start_lsn: Lsn,
         ancestor: Option<Arc<Timeline>>,
-        last_aux_file_policy: RuntimeAuxFilePolicy,
+        last_aux_file_policy: Option<AuxFilePolicy>,
     ) -> anyhow::Result<UninitializedTimeline> {
         let tenant_shard_id = self.tenant_shard_id;
 
diff --git a/pageserver/src/tenant/config.rs b/pageserver/src/tenant/config.rs
index c328ee8e63b4..e861617f3b86 100644
--- a/pageserver/src/tenant/config.rs
+++ b/pageserver/src/tenant/config.rs
@@ -11,7 +11,7 @@
 use anyhow::bail;
 use pageserver_api::models::CompactionAlgorithm;
 use pageserver_api::models::EvictionPolicy;
-use pageserver_api::models::SwitchAuxFilePolicy;
+use pageserver_api::models::AuxFilePolicy;
 use pageserver_api::models::{self, ThrottleConfig};
 use pageserver_api::shard::{ShardCount, ShardIdentity, ShardNumber, ShardStripeSize};
 use serde::de::IntoDeserializer;
@@ -375,7 +375,7 @@ pub struct TenantConf {
     /// 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: SwitchAuxFilePolicy,
+    pub switch_aux_file_policy: AuxFilePolicy,
 }
 
 /// Same as TenantConf, but this struct preserves the information about
@@ -474,7 +474,7 @@ pub struct TenantConfOpt {
 
     #[serde(skip_serializing_if = "Option::is_none")]
     #[serde(default)]
-    pub switch_aux_file_policy: Option<SwitchAuxFilePolicy>,
+    pub switch_aux_file_policy: Option<AuxFilePolicy>,
 }
 
 impl TenantConfOpt {
@@ -576,7 +576,7 @@ impl Default for TenantConf {
             lazy_slru_download: false,
             timeline_get_throttle: crate::tenant::throttle::Config::disabled(),
             image_layer_creation_check_threshold: DEFAULT_IMAGE_LAYER_CREATION_CHECK_THRESHOLD,
-            switch_aux_file_policy: SwitchAuxFilePolicy::V1,
+            switch_aux_file_policy: AuxFilePolicy::V1,
         }
     }
 }
diff --git a/pageserver/src/tenant/remote_timeline_client.rs b/pageserver/src/tenant/remote_timeline_client.rs
index 2ab59f90c3d2..ff26f8cc2ec2 100644
--- a/pageserver/src/tenant/remote_timeline_client.rs
+++ b/pageserver/src/tenant/remote_timeline_client.rs
@@ -189,7 +189,7 @@ use camino::Utf8Path;
 use chrono::{NaiveDateTime, Utc};
 
 pub(crate) use download::download_initdb_tar_zst;
-use pageserver_api::models::RuntimeAuxFilePolicy;
+use pageserver_api::models::AuxFilePolicy;
 use pageserver_api::shard::{ShardIndex, TenantShardId};
 use scopeguard::ScopeGuard;
 use tokio_util::sync::CancellationToken;
@@ -613,7 +613,7 @@ impl RemoteTimelineClient {
     /// Launch an index-file upload operation in the background, with only aux_file_policy flag updated.
     pub(crate) fn schedule_index_upload_for_aux_file_policy_update(
         self: &Arc<Self>,
-        last_aux_file_policy: RuntimeAuxFilePolicy,
+        last_aux_file_policy: Option<AuxFilePolicy>,
     ) -> anyhow::Result<()> {
         let mut guard = self.upload_queue.lock().unwrap();
         let upload_queue = guard.initialized_mut()?;
diff --git a/pageserver/src/tenant/remote_timeline_client/index.rs b/pageserver/src/tenant/remote_timeline_client/index.rs
index bd916b1a84db..b57c0e2fc65e 100644
--- a/pageserver/src/tenant/remote_timeline_client/index.rs
+++ b/pageserver/src/tenant/remote_timeline_client/index.rs
@@ -5,7 +5,7 @@
 use std::collections::HashMap;
 
 use chrono::NaiveDateTime;
-use pageserver_api::models::RuntimeAuxFilePolicy;
+use pageserver_api::models::AuxFilePolicy;
 use serde::{Deserialize, Serialize};
 use utils::id::TimelineId;
 
@@ -90,11 +90,11 @@ pub struct IndexPart {
     #[serde(default)]
     pub(crate) lineage: Lineage,
 
-    // Added in version 6. None == RuntimeAuxFilePolicy::Unspecified. Use Option wrapper to keep forward compatibility.
+    // Added in version 6. None == AuxFilePolicy::Unspecified. Use Option wrapper to keep forward compatibility.
     // This flag is controlled by TenantConf::switch_aux_file_policy. When the first aux file gets written, the aux file
     // mode will be persisted in `index_part.json`, and `switch_aux_file_policy` will be ignored.
     #[serde(skip_serializing_if = "Option::is_none", default)]
-    pub(crate) last_aux_file_policy: Option<RuntimeAuxFilePolicy>,
+    pub(crate) last_aux_file_policy: Option<AuxFilePolicy>,
 }
 
 impl IndexPart {
@@ -121,7 +121,7 @@ impl IndexPart {
         disk_consistent_lsn: Lsn,
         metadata: TimelineMetadata,
         lineage: Lineage,
-        last_aux_file_policy: RuntimeAuxFilePolicy,
+        last_aux_file_policy: Option<AuxFilePolicy>,
     ) -> Self {
         let layer_metadata = layers_and_metadata
             .iter()
@@ -135,7 +135,7 @@ impl IndexPart {
             metadata,
             deleted_at: None,
             lineage,
-            last_aux_file_policy: Some(last_aux_file_policy),
+            last_aux_file_policy,
         }
     }
 
@@ -165,13 +165,12 @@ impl IndexPart {
             example_metadata.disk_consistent_lsn(),
             example_metadata,
             Default::default(),
-            RuntimeAuxFilePolicy::V1,
+            Some(AuxFilePolicy::V1),
         )
     }
 
-    pub(crate) fn last_aux_file_policy(&self) -> RuntimeAuxFilePolicy {
+    pub(crate) fn last_aux_file_policy(&self) -> Option<AuxFilePolicy> {
         self.last_aux_file_policy
-            .unwrap_or(RuntimeAuxFilePolicy::Unspecified)
     }
 }
 
@@ -588,7 +587,7 @@ mod tests {
                 reparenting_history: vec![TimelineId::from_str("e1bfd8c633d713d279e6fcd2bcc15b6d").unwrap()],
                 original_ancestor: Some((TimelineId::from_str("e2bfd8c633d713d279e6fcd2bcc15b6d").unwrap(), Lsn::from_str("0/15A7618").unwrap(), parse_naive_datetime("2024-05-07T18:52:36.322426563"))),
             },
-            last_aux_file_policy: Some(RuntimeAuxFilePolicy::V2),
+            last_aux_file_policy: Some(AuxFilePolicy::V2),
         };
 
         let part = IndexPart::from_s3_bytes(example.as_bytes()).unwrap();
diff --git a/pageserver/src/tenant/timeline.rs b/pageserver/src/tenant/timeline.rs
index 75290f66078c..3d0c4600b6fc 100644
--- a/pageserver/src/tenant/timeline.rs
+++ b/pageserver/src/tenant/timeline.rs
@@ -23,9 +23,9 @@ use pageserver_api::{
     },
     keyspace::{KeySpaceAccum, SparseKeyPartitioning},
     models::{
-        AtomicRuntimeAuxFilePolicy, CompactionAlgorithm, DownloadRemoteLayersTaskInfo,
+        AtomicAuxFilePolicy, CompactionAlgorithm, DownloadRemoteLayersTaskInfo,
         DownloadRemoteLayersTaskSpawnRequest, EvictionPolicy, InMemoryLayerInfo, LayerMapInfo,
-        RuntimeAuxFilePolicy, SwitchAuxFilePolicy, TimelineState,
+        AuxFilePolicy, TimelineState,
     },
     reltag::BlockNumber,
     shard::{ShardIdentity, ShardNumber, TenantShardId},
@@ -417,7 +417,7 @@ pub struct Timeline {
     pub(crate) aux_file_size_estimator: AuxFileSizeEstimator,
 
     /// Indicate whether aux file v2 storage is enabled.
-    pub last_aux_file_policy: AtomicRuntimeAuxFilePolicy,
+    pub(crate) last_aux_file_policy: AtomicAuxFilePolicy,
 }
 
 pub struct WalReceiverInfo {
@@ -2015,7 +2015,7 @@ const REPARTITION_FREQ_IN_CHECKPOINT_DISTANCE: u64 = 10;
 
 // Private functions
 impl Timeline {
-    pub(crate) fn get_switch_aux_file_policy(&self) -> SwitchAuxFilePolicy {
+    pub(crate) fn get_switch_aux_file_policy(&self) -> AuxFilePolicy {
         let tenant_conf = self.tenant_conf.load();
         tenant_conf
             .tenant_conf
@@ -2153,7 +2153,7 @@ impl Timeline {
         resources: TimelineResources,
         pg_version: u32,
         state: TimelineState,
-        aux_file_policy: RuntimeAuxFilePolicy,
+        aux_file_policy: Option<AuxFilePolicy>,
         cancel: CancellationToken,
     ) -> Arc<Self> {
         let disk_consistent_lsn = metadata.disk_consistent_lsn();
@@ -2279,7 +2279,7 @@ impl Timeline {
 
                 aux_file_size_estimator: AuxFileSizeEstimator::new(aux_file_metrics),
 
-                last_aux_file_policy: AtomicRuntimeAuxFilePolicy::new(aux_file_policy),
+                last_aux_file_policy: AtomicAuxFilePolicy::new(aux_file_policy),
             };
             result.repartition_threshold =
                 result.get_checkpoint_distance() / REPARTITION_FREQ_IN_CHECKPOINT_DISTANCE;
diff --git a/pageserver/src/tenant/timeline/delete.rs b/pageserver/src/tenant/timeline/delete.rs
index dcbc9aadf03f..aef95cb9eaef 100644
--- a/pageserver/src/tenant/timeline/delete.rs
+++ b/pageserver/src/tenant/timeline/delete.rs
@@ -4,10 +4,7 @@ use std::{
 };
 
 use anyhow::Context;
-use pageserver_api::{
-    models::{RuntimeAuxFilePolicy, TimelineState},
-    shard::TenantShardId,
-};
+use pageserver_api::{models::TimelineState, shard::TenantShardId};
 use tokio::sync::OwnedMutexGuard;
 use tracing::{error, info, instrument, Instrument};
 use utils::{crashsafe, fs_ext, id::TimelineId};
@@ -282,7 +279,7 @@ impl DeleteTimelineFlow {
                 // Thus we need to skip the validation here.
                 CreateTimelineCause::Delete,
                 // Aux file policy is not needed for deletion, assuming deletion does not read aux keyspace
-                RuntimeAuxFilePolicy::Unspecified,
+                None,
             )
             .context("create_timeline_struct")?;
 
diff --git a/pageserver/src/tenant/upload_queue.rs b/pageserver/src/tenant/upload_queue.rs
index f4f9c286f129..4439344f037e 100644
--- a/pageserver/src/tenant/upload_queue.rs
+++ b/pageserver/src/tenant/upload_queue.rs
@@ -8,7 +8,7 @@ use std::collections::{HashMap, VecDeque};
 use std::fmt::Debug;
 
 use chrono::NaiveDateTime;
-use pageserver_api::models::RuntimeAuxFilePolicy;
+use pageserver_api::models::AuxFilePolicy;
 use std::sync::Arc;
 use tracing::info;
 use utils::lsn::AtomicLsn;
@@ -61,7 +61,8 @@ pub(crate) struct UploadQueueInitialized {
     /// Part of the flattened "next" `index_part.json`.
     pub(crate) latest_lineage: Lineage,
 
-    pub(crate) last_aux_file_policy: RuntimeAuxFilePolicy,
+    /// The last aux file policy used on this timeline.
+    pub(crate) last_aux_file_policy: Option<AuxFilePolicy>,
 
     /// `disk_consistent_lsn` from the last metadata file that was successfully
     /// uploaded. `Lsn(0)` if nothing was uploaded yet.
@@ -192,7 +193,7 @@ impl UploadQueue {
             dangling_files: HashMap::new(),
             shutting_down: false,
             shutdown_ready: Arc::new(tokio::sync::Semaphore::new(0)),
-            last_aux_file_policy: RuntimeAuxFilePolicy::Unspecified,
+            last_aux_file_policy: None,
         };
 
         *self = UploadQueue::Initialized(state);
diff --git a/test_runner/regress/test_aux_files.py b/test_runner/regress/test_aux_files.py
index 139d49c9f23d..08d662617837 100644
--- a/test_runner/regress/test_aux_files.py
+++ b/test_runner/regress/test_aux_files.py
@@ -20,10 +20,7 @@ def test_aux_v2_config_switch(neon_simple_env: NeonEnv, vanilla_pg):
         client.set_tenant_config(tenant_id, tenant_config)
         # aux file v2 is enabled on the write path
         assert (
-            client.timeline_detail(tenant_id=tenant_id, timeline_id=timeline_id)[
-                "last_aux_file_policy"
-            ]
-            == "Unspecified"
+            "last_aux_file_policy" not in client.timeline_detail(tenant_id=tenant_id, timeline_id=timeline_id)
         )
     pg_conn = endpoint.connect()
     cur = pg_conn.cursor()

From 5ed4b7ada2830d5b87205eb59d2b90713ca6c95b Mon Sep 17 00:00:00 2001
From: Alex Chi Z <chi@neon.tech>
Date: Wed, 15 May 2024 14:18:33 -0400
Subject: [PATCH 06/20] resolve comments for e2e test

Signed-off-by: Alex Chi Z <chi@neon.tech>
---
 test_runner/regress/test_aux_files.py | 35 +++++++++++----------------
 1 file changed, 14 insertions(+), 21 deletions(-)

diff --git a/test_runner/regress/test_aux_files.py b/test_runner/regress/test_aux_files.py
index 08d662617837..e8856b6d572f 100644
--- a/test_runner/regress/test_aux_files.py
+++ b/test_runner/regress/test_aux_files.py
@@ -1,27 +1,26 @@
 from fixtures.log_helper import log
 from fixtures.neon_fixtures import (
-    NeonEnv,
+    NeonEnvBuilder,
     logical_replication_sync,
 )
 
 
-def test_aux_v2_config_switch(neon_simple_env: NeonEnv, vanilla_pg):
-    env = neon_simple_env
+def test_aux_v2_config_switch(neon_env_builder: NeonEnvBuilder, vanilla_pg):
+    env = neon_env_builder.init_start()
+    endpoint = env.endpoints.create_start("main")
+    client = env.pageserver.http_client()
 
     tenant_id = env.initial_tenant
-    timeline_id = env.neon_cli.create_branch("test_aux_v2_config_switch", "empty")
-    endpoint = env.endpoints.create_start(
-        "test_aux_v2_config_switch", config_lines=["log_statement=all"]
+    timeline_id = env.initial_timeline
+
+    tenant_config = client.tenant_config(tenant_id).effective_config
+    tenant_config["switch_aux_file_policy"] = "V2"
+    client.set_tenant_config(tenant_id, tenant_config)
+    # aux file v2 is enabled on the write path, so for now, it should be unset (or null)
+    assert "last_aux_file_policy" not in client.timeline_detail(
+        tenant_id=tenant_id, timeline_id=timeline_id
     )
 
-    with env.pageserver.http_client() as client:
-        tenant_config = client.tenant_config(tenant_id).effective_config
-        tenant_config["switch_aux_file_policy"] = "V2"
-        client.set_tenant_config(tenant_id, tenant_config)
-        # aux file v2 is enabled on the write path
-        assert (
-            "last_aux_file_policy" not in client.timeline_detail(tenant_id=tenant_id, timeline_id=timeline_id)
-        )
     pg_conn = endpoint.connect()
     cur = pg_conn.cursor()
 
@@ -49,12 +48,7 @@ def test_aux_v2_config_switch(neon_simple_env: NeonEnv, vanilla_pg):
 
     with env.pageserver.http_client() as client:
         # aux file v2 flag should be enabled at this point
-        assert (
-            client.timeline_detail(tenant_id=tenant_id, timeline_id=timeline_id)[
-                "last_aux_file_policy"
-            ]
-            == "V2"
-        )
+        assert client.timeline_detail(tenant_id, timeline_id)["last_aux_file_policy"] == "V2"
     with env.pageserver.http_client() as client:
         tenant_config = client.tenant_config(tenant_id).effective_config
         tenant_config["switch_aux_file_policy"] = "V1"
@@ -75,4 +69,3 @@ def test_aux_v2_config_switch(neon_simple_env: NeonEnv, vanilla_pg):
             ]
             == "V2"
         )
-    # TODO(chi): test with timeline detach?

From 72cd2c389a6b8e801aabca02940330d3644ca644 Mon Sep 17 00:00:00 2001
From: Alex Chi Z <chi@neon.tech>
Date: Wed, 15 May 2024 14:27:46 -0400
Subject: [PATCH 07/20] move migration path to unit tests

Signed-off-by: Alex Chi Z <chi@neon.tech>
---
 libs/pageserver_api/src/models.rs   | 68 +++++++++++++++++++++++++++--
 pageserver/src/pgdatadir_mapping.rs | 29 ++++++------
 2 files changed, 77 insertions(+), 20 deletions(-)

diff --git a/libs/pageserver_api/src/models.rs b/libs/pageserver_api/src/models.rs
index e61215cb692e..68928d8ed60f 100644
--- a/libs/pageserver_api/src/models.rs
+++ b/libs/pageserver_api/src/models.rs
@@ -329,6 +329,15 @@ 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)
+        )
+    }
+}
+
 /// The aux file policy memory flag. Users can store `Option<AuxFilePolicy>` into this atomic flag. 0 == unspecified.
 pub struct AtomicAuxFilePolicy(AtomicUsize);
 
@@ -377,10 +386,6 @@ impl AuxFilePolicy {
     pub fn from_usize(this: usize) -> Self {
         Self::try_from_usize(this).unwrap()
     }
-
-    pub fn to_runtime_policy(&self) -> AuxFilePolicy {
-        AuxFilePolicy::from_usize(self.to_usize())
-    }
 }
 
 impl FromStr for AuxFilePolicy {
@@ -1520,4 +1525,59 @@ mod tests {
             assert_eq!(actual, expected, "example on {line}");
         }
     }
+
+    #[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
+        ));
+    }
 }
diff --git a/pageserver/src/pgdatadir_mapping.rs b/pageserver/src/pgdatadir_mapping.rs
index c7da9211baa6..7324fe7e77fe 100644
--- a/pageserver/src/pgdatadir_mapping.rs
+++ b/pageserver/src/pgdatadir_mapping.rs
@@ -24,7 +24,7 @@ use pageserver_api::key::{
     AUX_FILES_KEY, CHECKPOINT_KEY, CONTROLFILE_KEY, DBDIR_KEY, TWOPHASEDIR_KEY,
 };
 use pageserver_api::keyspace::SparseKeySpace;
-use pageserver_api::models::{AuxFilePolicy, };
+use pageserver_api::models::AuxFilePolicy;
 use pageserver_api::reltag::{BlockNumber, RelTag, SlruKind};
 use postgres_ffi::relfile_utils::{FSM_FORKNUM, VISIBILITYMAP_FORKNUM};
 use postgres_ffi::BLCKSZ;
@@ -720,9 +720,7 @@ impl Timeline {
     ) -> Result<HashMap<String, Bytes>, PageReconstructError> {
         let current_policy = self.last_aux_file_policy.load();
         match current_policy {
-            Some(AuxFilePolicy::V1) | None => {
-                self.list_aux_files_v1(lsn, ctx).await
-            }
+            Some(AuxFilePolicy::V1) | None => self.list_aux_files_v1(lsn, ctx).await,
             Some(AuxFilePolicy::V2) => self.list_aux_files_v2(lsn, ctx).await,
             Some(AuxFilePolicy::CrossValidation) => {
                 let v1_result = self.list_aux_files_v1(lsn, ctx).await;
@@ -1479,19 +1477,18 @@ impl<'a> DatadirModification<'a> {
             // Allowed switch path:
             // * no aux files -> v1/v2/cross-validation
             // * cross-validation->v2
-            match (current_policy, switch_policy) {
-                (None, _)
-                | (Some(AuxFilePolicy::CrossValidation), AuxFilePolicy::V2) => {
-                    let new_policy = switch_policy.to_runtime_policy();
-                    self.tline.last_aux_file_policy.store(Some(new_policy));
-                    info!("switching to aux file policy {:?}", new_policy);
-                    if let Some(ref remote_client) = self.tline.remote_client {
-                        remote_client
-                            .schedule_index_upload_for_aux_file_policy_update(Some(new_policy))?;
-                    }
-                    new_policy
+            if AuxFilePolicy::is_valid_migration_path(current_policy, switch_policy) {
+                self.tline.last_aux_file_policy.store(Some(switch_policy));
+                info!("switching to aux file policy {:?}", switch_policy);
+                if let Some(ref remote_client) = self.tline.remote_client {
+                    remote_client
+                        .schedule_index_upload_for_aux_file_policy_update(Some(switch_policy))?;
                 }
-                (Some(current_policy), _) => current_policy,
+                switch_policy
+            } else {
+                // This branch handles non-valid migration path, and the case that switch_policy == current_policy.
+                // And actually, because the migration path always allow unspecified -> *, this unwrap_or will never be hit.
+                current_policy.unwrap_or(AuxFilePolicy::V1)
             }
         };
 

From 2f3acac4eac42a4ca00f09aed79d05a29080689f Mon Sep 17 00:00:00 2001
From: Alex Chi Z <chi@neon.tech>
Date: Wed, 15 May 2024 14:28:35 -0400
Subject: [PATCH 08/20] revert wait_completion pub(crate)

Signed-off-by: Alex Chi Z <chi@neon.tech>
---
 pageserver/src/tenant/remote_timeline_client.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/pageserver/src/tenant/remote_timeline_client.rs b/pageserver/src/tenant/remote_timeline_client.rs
index ff26f8cc2ec2..9d899e5be153 100644
--- a/pageserver/src/tenant/remote_timeline_client.rs
+++ b/pageserver/src/tenant/remote_timeline_client.rs
@@ -954,7 +954,7 @@ impl RemoteTimelineClient {
     }
 
     /// Wait for all previously scheduled uploads/deletions to complete
-    pub(crate) async fn wait_completion(self: &Arc<Self>) -> anyhow::Result<()> {
+    async fn wait_completion(self: &Arc<Self>) -> anyhow::Result<()> {
         let receiver = {
             let mut guard = self.upload_queue.lock().unwrap();
             let upload_queue = guard.initialized_mut()?;

From c369b89356a1b9fd3986787583aa514e1d4e4db3 Mon Sep 17 00:00:00 2001
From: Alex Chi Z <chi@neon.tech>
Date: Wed, 15 May 2024 14:42:08 -0400
Subject: [PATCH 09/20] unit test for aux flags on branching

Signed-off-by: Alex Chi Z <chi@neon.tech>
---
 control_plane/src/pageserver.rs               |  2 +-
 pageserver/src/tenant.rs                      | 27 +++++++++++++++++++
 pageserver/src/tenant/config.rs               |  2 +-
 .../src/tenant/remote_timeline_client.rs      |  4 +--
 pageserver/src/tenant/timeline.rs             |  4 +--
 5 files changed, 33 insertions(+), 6 deletions(-)

diff --git a/control_plane/src/pageserver.rs b/control_plane/src/pageserver.rs
index 4b2b0348290d..5a8476369769 100644
--- a/control_plane/src/pageserver.rs
+++ b/control_plane/src/pageserver.rs
@@ -17,7 +17,7 @@ use anyhow::{bail, Context};
 use camino::Utf8PathBuf;
 use futures::SinkExt;
 use pageserver_api::models::{
-    self, LocationConfig, ShardParameters, AuxFilePolicy, TenantHistorySize, TenantInfo,
+    self, AuxFilePolicy, LocationConfig, ShardParameters, TenantHistorySize, TenantInfo,
     TimelineInfo,
 };
 use pageserver_api::shard::TenantShardId;
diff --git a/pageserver/src/tenant.rs b/pageserver/src/tenant.rs
index b605961b4590..26da84be559d 100644
--- a/pageserver/src/tenant.rs
+++ b/pageserver/src/tenant.rs
@@ -5644,4 +5644,31 @@ mod tests {
 
         Ok(())
     }
+
+    #[tokio::test]
+    async fn test_aux_flag_branch() -> anyhow::Result<()> {
+        let mut harness = TenantHarness::create("aux_flag_branch")?;
+        harness.tenant_conf.switch_aux_file_policy = AuxFilePolicy::V2;
+        let (tenant, ctx) = harness.load().await;
+        let tline: Arc<Timeline> = tenant
+            .create_test_timeline(TIMELINE_ID, Lsn(0x08), DEFAULT_PG_VERSION, &ctx)
+            .await?;
+        assert_eq!(tline.last_aux_file_policy.load(), None);
+        {
+            let mut modification = tline.begin_modification(Lsn(0x10));
+            modification
+                .put_file("pg_logical/mappings/test", b"test", &ctx)
+                .await?;
+            modification.commit(&ctx).await?;
+        }
+        assert_eq!(tline.last_aux_file_policy.load(), Some(AuxFilePolicy::V2));
+        let create_guard = tenant
+            .create_timeline_create_guard(NEW_TIMELINE_ID)
+            .unwrap();
+        let tline2 = tenant
+            .branch_timeline_impl(&tline, NEW_TIMELINE_ID, None, create_guard, &ctx)
+            .await?;
+        assert_eq!(tline2.last_aux_file_policy.load(), Some(AuxFilePolicy::V2));
+        Ok(())
+    }
 }
diff --git a/pageserver/src/tenant/config.rs b/pageserver/src/tenant/config.rs
index e861617f3b86..d96766ed05b9 100644
--- a/pageserver/src/tenant/config.rs
+++ b/pageserver/src/tenant/config.rs
@@ -9,9 +9,9 @@
 //! may lead to a data loss.
 //!
 use anyhow::bail;
+use pageserver_api::models::AuxFilePolicy;
 use pageserver_api::models::CompactionAlgorithm;
 use pageserver_api::models::EvictionPolicy;
-use pageserver_api::models::AuxFilePolicy;
 use pageserver_api::models::{self, ThrottleConfig};
 use pageserver_api::shard::{ShardCount, ShardIdentity, ShardNumber, ShardStripeSize};
 use serde::de::IntoDeserializer;
diff --git a/pageserver/src/tenant/remote_timeline_client.rs b/pageserver/src/tenant/remote_timeline_client.rs
index 9d899e5be153..b1e05e3aabb3 100644
--- a/pageserver/src/tenant/remote_timeline_client.rs
+++ b/pageserver/src/tenant/remote_timeline_client.rs
@@ -954,8 +954,8 @@ impl RemoteTimelineClient {
     }
 
     /// Wait for all previously scheduled uploads/deletions to complete
-    async fn wait_completion(self: &Arc<Self>) -> anyhow::Result<()> {
-        let receiver = {
+    pub(crate) async fn wait_completion(self: &Arc<Self>) -> anyhow::Result<()> {
+        let receiver: tokio::sync::watch::Receiver<()> = {
             let mut guard = self.upload_queue.lock().unwrap();
             let upload_queue = guard.initialized_mut()?;
             self.schedule_barrier0(upload_queue)
diff --git a/pageserver/src/tenant/timeline.rs b/pageserver/src/tenant/timeline.rs
index 3d0c4600b6fc..8cb1058a9d15 100644
--- a/pageserver/src/tenant/timeline.rs
+++ b/pageserver/src/tenant/timeline.rs
@@ -23,9 +23,9 @@ use pageserver_api::{
     },
     keyspace::{KeySpaceAccum, SparseKeyPartitioning},
     models::{
-        AtomicAuxFilePolicy, CompactionAlgorithm, DownloadRemoteLayersTaskInfo,
+        AtomicAuxFilePolicy, AuxFilePolicy, CompactionAlgorithm, DownloadRemoteLayersTaskInfo,
         DownloadRemoteLayersTaskSpawnRequest, EvictionPolicy, InMemoryLayerInfo, LayerMapInfo,
-        AuxFilePolicy, TimelineState,
+        TimelineState,
     },
     reltag::BlockNumber,
     shard::{ShardIdentity, ShardNumber, TenantShardId},

From 071ce5d3e42fabc1b91ddc8c7585e55bd77637d9 Mon Sep 17 00:00:00 2001
From: Alex Chi Z <chi@neon.tech>
Date: Wed, 15 May 2024 14:51:06 -0400
Subject: [PATCH 10/20] mv set last_aux_file_policy in timeline

Signed-off-by: Alex Chi Z <chi@neon.tech>
---
 pageserver/src/pgdatadir_mapping.rs | 7 +++----
 pageserver/src/tenant.rs            | 4 ++++
 pageserver/src/tenant/timeline.rs   | 5 -----
 3 files changed, 7 insertions(+), 9 deletions(-)

diff --git a/pageserver/src/pgdatadir_mapping.rs b/pageserver/src/pgdatadir_mapping.rs
index 7324fe7e77fe..f02f32edd9b2 100644
--- a/pageserver/src/pgdatadir_mapping.rs
+++ b/pageserver/src/pgdatadir_mapping.rs
@@ -1480,10 +1480,9 @@ impl<'a> DatadirModification<'a> {
             if AuxFilePolicy::is_valid_migration_path(current_policy, switch_policy) {
                 self.tline.last_aux_file_policy.store(Some(switch_policy));
                 info!("switching to aux file policy {:?}", switch_policy);
-                if let Some(ref remote_client) = self.tline.remote_client {
-                    remote_client
-                        .schedule_index_upload_for_aux_file_policy_update(Some(switch_policy))?;
-                }
+                self.tline
+                    .remote_client
+                    .schedule_index_upload_for_aux_file_policy_update(Some(switch_policy))?;
                 switch_policy
             } else {
                 // This branch handles non-valid migration path, and the case that switch_policy == current_policy.
diff --git a/pageserver/src/tenant.rs b/pageserver/src/tenant.rs
index 009e3bdc160e..4467ba949a51 100644
--- a/pageserver/src/tenant.rs
+++ b/pageserver/src/tenant.rs
@@ -558,6 +558,10 @@ impl Tenant {
 
         if let Some(index_part) = index_part.as_ref() {
             timeline.remote_client.init_upload_queue(index_part)?;
+
+            timeline
+                .last_aux_file_policy
+                .store(index_part.last_aux_file_policy());
         } else {
             // No data on the remote storage, but we have local metadata file. We can end up
             // here with timeline_create being interrupted before finishing index part upload.
diff --git a/pageserver/src/tenant/timeline.rs b/pageserver/src/tenant/timeline.rs
index fceb6cdc83e9..1fb1928079b1 100644
--- a/pageserver/src/tenant/timeline.rs
+++ b/pageserver/src/tenant/timeline.rs
@@ -2413,11 +2413,6 @@ impl Timeline {
         let shard = self.get_shard_index();
         let this = self.myself.upgrade().expect("&self method holds the arc");
 
-        if let Some(ref index_part) = index_part {
-            self.last_aux_file_policy
-                .store(index_part.last_aux_file_policy());
-        }
-
         let (loaded_layers, needs_cleanup, total_physical_size) = tokio::task::spawn_blocking({
             move || {
                 let _g = span.entered();

From a2c91d68916ff64ea05f8b199f1e3efb14ec824c Mon Sep 17 00:00:00 2001
From: Alex Chi Z <chi@neon.tech>
Date: Wed, 15 May 2024 15:20:45 -0400
Subject: [PATCH 11/20] fix tests

Signed-off-by: Alex Chi Z <chi@neon.tech>
---
 test_runner/regress/test_aux_files.py | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/test_runner/regress/test_aux_files.py b/test_runner/regress/test_aux_files.py
index e8856b6d572f..be9c41a8679b 100644
--- a/test_runner/regress/test_aux_files.py
+++ b/test_runner/regress/test_aux_files.py
@@ -17,8 +17,9 @@ def test_aux_v2_config_switch(neon_env_builder: NeonEnvBuilder, vanilla_pg):
     tenant_config["switch_aux_file_policy"] = "V2"
     client.set_tenant_config(tenant_id, tenant_config)
     # aux file v2 is enabled on the write path, so for now, it should be unset (or null)
-    assert "last_aux_file_policy" not in client.timeline_detail(
-        tenant_id=tenant_id, timeline_id=timeline_id
+    assert (
+        client.timeline_detail(tenant_id=tenant_id, timeline_id=timeline_id)["last_aux_file_policy"]
+        is None
     )
 
     pg_conn = endpoint.connect()

From 9dfb8b9ff45d0a12ea5bba91180173b809e23a1f Mon Sep 17 00:00:00 2001
From: Alex Chi Z <chi@neon.tech>
Date: Thu, 16 May 2024 11:26:42 -0400
Subject: [PATCH 12/20] use default for queue uploads

Signed-off-by: Alex Chi Z <chi@neon.tech>
---
 pageserver/src/tenant/upload_queue.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/pageserver/src/tenant/upload_queue.rs b/pageserver/src/tenant/upload_queue.rs
index 4439344f037e..c0cc8f3124de 100644
--- a/pageserver/src/tenant/upload_queue.rs
+++ b/pageserver/src/tenant/upload_queue.rs
@@ -193,7 +193,7 @@ impl UploadQueue {
             dangling_files: HashMap::new(),
             shutting_down: false,
             shutdown_ready: Arc::new(tokio::sync::Semaphore::new(0)),
-            last_aux_file_policy: None,
+            last_aux_file_policy: Default::default(),
         };
 
         *self = UploadQueue::Initialized(state);

From 3b1b74e06d2c52534cd3dad6bbec03c73a0194ef Mon Sep 17 00:00:00 2001
From: Alex Chi Z <chi@neon.tech>
Date: Thu, 16 May 2024 11:27:44 -0400
Subject: [PATCH 13/20] rename test to test_branch_copies_aux_file_flag

Signed-off-by: Alex Chi Z <chi@neon.tech>
---
 pageserver/src/tenant.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/pageserver/src/tenant.rs b/pageserver/src/tenant.rs
index 4467ba949a51..cf535472c507 100644
--- a/pageserver/src/tenant.rs
+++ b/pageserver/src/tenant.rs
@@ -5617,7 +5617,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn test_aux_flag_branch() -> anyhow::Result<()> {
+    async fn test_branch_copies_aux_file_flag() -> anyhow::Result<()> {
         let mut harness = TenantHarness::create("aux_flag_branch")?;
         harness.tenant_conf.switch_aux_file_policy = AuxFilePolicy::V2;
         let (tenant, ctx) = harness.load().await;

From eb8f9a39920dfff104fc8f61a6117e7d4fdcd6c3 Mon Sep 17 00:00:00 2001
From: Alex Chi Z <chi@neon.tech>
Date: Thu, 16 May 2024 11:28:09 -0400
Subject: [PATCH 14/20] rm unnecessary type def

Signed-off-by: Alex Chi Z <chi@neon.tech>
---
 pageserver/src/tenant/remote_timeline_client.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/pageserver/src/tenant/remote_timeline_client.rs b/pageserver/src/tenant/remote_timeline_client.rs
index 097309fca7b9..609a3d4dcd3b 100644
--- a/pageserver/src/tenant/remote_timeline_client.rs
+++ b/pageserver/src/tenant/remote_timeline_client.rs
@@ -955,7 +955,7 @@ impl RemoteTimelineClient {
 
     /// Wait for all previously scheduled uploads/deletions to complete
     pub(crate) async fn wait_completion(self: &Arc<Self>) -> anyhow::Result<()> {
-        let receiver: tokio::sync::watch::Receiver<()> = {
+        let receiver = {
             let mut guard = self.upload_queue.lock().unwrap();
             let upload_queue = guard.initialized_mut()?;
             self.schedule_barrier0(upload_queue)

From 37b83526e832df80dec728214b3c2dbf85886bff Mon Sep 17 00:00:00 2001
From: Alex Chi Z <chi@neon.tech>
Date: Thu, 16 May 2024 11:33:25 -0400
Subject: [PATCH 15/20] update indexpart comments

Signed-off-by: Alex Chi Z <chi@neon.tech>
---
 pageserver/src/tenant/remote_timeline_client/index.rs | 10 +++++++---
 1 file changed, 7 insertions(+), 3 deletions(-)

diff --git a/pageserver/src/tenant/remote_timeline_client/index.rs b/pageserver/src/tenant/remote_timeline_client/index.rs
index b57c0e2fc65e..88d4daad5ec8 100644
--- a/pageserver/src/tenant/remote_timeline_client/index.rs
+++ b/pageserver/src/tenant/remote_timeline_client/index.rs
@@ -90,9 +90,13 @@ pub struct IndexPart {
     #[serde(default)]
     pub(crate) lineage: Lineage,
 
-    // Added in version 6. None == AuxFilePolicy::Unspecified. Use Option wrapper to keep forward compatibility.
-    // This flag is controlled by TenantConf::switch_aux_file_policy. When the first aux file gets written, the aux file
-    // mode will be persisted in `index_part.json`, and `switch_aux_file_policy` will be ignored.
+    /// Describes the kind of aux files stored in the timeline.
+    /// 
+    /// The value is modified during file ingestion when the latest wanted value communicated via tenant config is applied if it is acceptable.
+    /// A V1 setting after V2 files have been committed is not accepted.
+    /// 
+    /// None means no aux files have been written to the storage before the point
+    /// when this flag is introduced.
     #[serde(skip_serializing_if = "Option::is_none", default)]
     pub(crate) last_aux_file_policy: Option<AuxFilePolicy>,
 }

From 38c6037974c4813feeacca6fc4d04bde8d3c091f Mon Sep 17 00:00:00 2001
From: Alex Chi Z <chi@neon.tech>
Date: Thu, 16 May 2024 12:36:55 -0400
Subject: [PATCH 16/20] Acq/Rel for aux file atomic flag

Signed-off-by: Alex Chi Z <chi@neon.tech>
---
 libs/pageserver_api/src/models.rs | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/libs/pageserver_api/src/models.rs b/libs/pageserver_api/src/models.rs
index 14f46f026630..ef994436b599 100644
--- a/libs/pageserver_api/src/models.rs
+++ b/libs/pageserver_api/src/models.rs
@@ -351,7 +351,7 @@ impl AtomicAuxFilePolicy {
     pub fn load(&self) -> Option<AuxFilePolicy> {
         // Using SeqCst memory order because this flag will be read from both write/read threads
         // and we want to guarantee the strongest consistency.
-        match self.0.load(std::sync::atomic::Ordering::SeqCst) {
+        match self.0.load(std::sync::atomic::Ordering::Acquire) {
             0 => None,
             other => Some(AuxFilePolicy::from_usize(other)),
         }
@@ -360,7 +360,7 @@ impl AtomicAuxFilePolicy {
     pub fn store(&self, policy: Option<AuxFilePolicy>) {
         self.0.store(
             policy.map(AuxFilePolicy::to_usize).unwrap_or_default(),
-            std::sync::atomic::Ordering::SeqCst,
+            std::sync::atomic::Ordering::Release,
         );
     }
 }

From e7203d91ae3c6cb25339a5a0106a9a265acc56b1 Mon Sep 17 00:00:00 2001
From: Alex Chi Z <chi@neon.tech>
Date: Thu, 16 May 2024 12:44:23 -0400
Subject: [PATCH 17/20] incorporate test cases from 7781

Co-Authored-By: Joonas Koivunen <joonas@neon.tech>
Signed-off-by: Alex Chi Z <chi@neon.tech>
---
 pageserver/src/pgdatadir_mapping.rs           |   2 +-
 pageserver/src/tenant.rs                      | 283 ++++++++++++++++--
 .../tenant/remote_timeline_client/index.rs    |   4 +-
 3 files changed, 269 insertions(+), 20 deletions(-)

diff --git a/pageserver/src/pgdatadir_mapping.rs b/pageserver/src/pgdatadir_mapping.rs
index f02f32edd9b2..23e43e8e9274 100644
--- a/pageserver/src/pgdatadir_mapping.rs
+++ b/pageserver/src/pgdatadir_mapping.rs
@@ -1479,10 +1479,10 @@ impl<'a> DatadirModification<'a> {
             // * cross-validation->v2
             if AuxFilePolicy::is_valid_migration_path(current_policy, switch_policy) {
                 self.tline.last_aux_file_policy.store(Some(switch_policy));
-                info!("switching to aux file policy {:?}", switch_policy);
                 self.tline
                     .remote_client
                     .schedule_index_upload_for_aux_file_policy_update(Some(switch_policy))?;
+                info!(current=?current_policy, next=?switch_policy, "switching aux file policy");
                 switch_policy
             } else {
                 // This branch handles non-valid migration path, and the case that switch_policy == current_policy.
diff --git a/pageserver/src/tenant.rs b/pageserver/src/tenant.rs
index cf535472c507..136b3d3e0f4d 100644
--- a/pageserver/src/tenant.rs
+++ b/pageserver/src/tenant.rs
@@ -5617,29 +5617,278 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn test_branch_copies_aux_file_flag() -> anyhow::Result<()> {
-        let mut harness = TenantHarness::create("aux_flag_branch")?;
-        harness.tenant_conf.switch_aux_file_policy = AuxFilePolicy::V2;
+    async fn test_branch_copies_dirty_aux_file_flag() {
+        let harness = TenantHarness::create("test_branch_copies_dirty_aux_file_flag").unwrap();
+
+        // the default aux file policy to switch is v1 if not set by the admins
+        assert_eq!(
+            harness.tenant_conf.switch_aux_file_policy,
+            AuxFilePolicy::V1
+        );
         let (tenant, ctx) = harness.load().await;
+
+        let mut lsn = Lsn(0x08);
+
         let tline: Arc<Timeline> = tenant
-            .create_test_timeline(TIMELINE_ID, Lsn(0x08), DEFAULT_PG_VERSION, &ctx)
-            .await?;
+            .create_test_timeline(TIMELINE_ID, lsn, DEFAULT_PG_VERSION, &ctx)
+            .await
+            .unwrap();
+
+        // no aux file is written at this point, so the persistent flag should be unset
         assert_eq!(tline.last_aux_file_policy.load(), None);
+
         {
-            let mut modification = tline.begin_modification(Lsn(0x10));
+            lsn += 8;
+            let mut modification = tline.begin_modification(lsn);
             modification
-                .put_file("pg_logical/mappings/test", b"test", &ctx)
-                .await?;
-            modification.commit(&ctx).await?;
+                .put_file("pg_logical/mappings/test1", b"first", &ctx)
+                .await
+                .unwrap();
+            modification.commit(&ctx).await.unwrap();
         }
-        assert_eq!(tline.last_aux_file_policy.load(), Some(AuxFilePolicy::V2));
-        let create_guard = tenant
-            .create_timeline_create_guard(NEW_TIMELINE_ID)
+
+        // there is no tenant manager to pass the configuration through, so lets mimic it
+        tenant.set_new_location_config(
+            AttachedTenantConf::try_from(LocationConf::attached_single(
+                TenantConfOpt {
+                    switch_aux_file_policy: Some(AuxFilePolicy::V2),
+                    ..Default::default()
+                },
+                tenant.generation,
+                &pageserver_api::models::ShardParameters::default(),
+            ))
+            .unwrap(),
+        );
+
+        assert_eq!(
+            tline.get_switch_aux_file_policy(),
+            AuxFilePolicy::V2,
+            "wanted state has been updated"
+        );
+        assert_eq!(
+            tline.last_aux_file_policy.load(),
+            Some(AuxFilePolicy::V1),
+            "aux file is written with switch_aux_file_policy unset (which is v1), so we should keep v1"
+        );
+
+        // we can still read the auxfile v1 before we ingest anything new
+        let files = tline.list_aux_files(lsn, &ctx).await.unwrap();
+        assert_eq!(
+            files.get("pg_logical/mappings/test1"),
+            Some(&bytes::Bytes::from_static(b"first"))
+        );
+
+        {
+            lsn += 8;
+            let mut modification = tline.begin_modification(lsn);
+            modification
+                .put_file("pg_logical/mappings/test2", b"second", &ctx)
+                .await
+                .unwrap();
+            modification.commit(&ctx).await.unwrap();
+        }
+
+        assert_eq!(
+            tline.last_aux_file_policy.load(),
+            Some(AuxFilePolicy::V1),
+            "keep v1 storage format on branching"
+        );
+
+        let files = tline.list_aux_files(lsn, &ctx).await.unwrap();
+        assert_eq!(
+            files.get("pg_logical/mappings/test2"),
+            Some(&bytes::Bytes::from_static(b"second"))
+        );
+
+        let child = tenant
+            .branch_timeline_test(&tline, NEW_TIMELINE_ID, Some(lsn), &ctx)
+            .await
             .unwrap();
-        let tline2 = tenant
-            .branch_timeline_impl(&tline, NEW_TIMELINE_ID, None, create_guard, &ctx)
-            .await?;
-        assert_eq!(tline2.last_aux_file_policy.load(), Some(AuxFilePolicy::V2));
-        Ok(())
+
+        // child copies the last flag even if that is not on remote storage yet
+        assert_eq!(child.get_switch_aux_file_policy(), AuxFilePolicy::V2);
+        assert_eq!(child.last_aux_file_policy.load(), Some(AuxFilePolicy::V1));
+
+        let files = child.list_aux_files(lsn, &ctx).await.unwrap();
+        assert_eq!(files.get("pg_logical/mappings/test1"), None);
+        assert_eq!(files.get("pg_logical/mappings/test2"), None);
+
+        // even if we crash here without flushing parent timeline with it's new
+        // last_aux_file_policy we are safe, because child was never meant to access ancestor's
+        // files. the ancestor can even switch back to V1 because of a migration safely.
+    }
+
+    #[tokio::test]
+    async fn aux_file_policy_switch() {
+        let mut harness = TenantHarness::create("aux_file_policy_switch").unwrap();
+        harness.tenant_conf.switch_aux_file_policy = AuxFilePolicy::CrossValidation; // set to cross-validation mode
+        let (tenant, ctx) = harness.load().await;
+
+        let mut lsn = Lsn(0x08);
+
+        let tline: Arc<Timeline> = tenant
+            .create_test_timeline(TIMELINE_ID, lsn, DEFAULT_PG_VERSION, &ctx)
+            .await
+            .unwrap();
+
+        assert_eq!(
+            tline.last_aux_file_policy.load(),
+            None,
+            "no aux file is written so it should be unset"
+        );
+
+        {
+            lsn += 8;
+            let mut modification = tline.begin_modification(lsn);
+            modification
+                .put_file("pg_logical/mappings/test1", b"first", &ctx)
+                .await
+                .unwrap();
+            modification.commit(&ctx).await.unwrap();
+        }
+
+        // there is no tenant manager to pass the configuration through, so lets mimic it
+        tenant.set_new_location_config(
+            AttachedTenantConf::try_from(LocationConf::attached_single(
+                TenantConfOpt {
+                    switch_aux_file_policy: Some(AuxFilePolicy::V2),
+                    ..Default::default()
+                },
+                tenant.generation,
+                &pageserver_api::models::ShardParameters::default(),
+            ))
+            .unwrap(),
+        );
+
+        assert_eq!(
+            tline.get_switch_aux_file_policy(),
+            AuxFilePolicy::V2,
+            "wanted state has been updated"
+        );
+        assert_eq!(
+            tline.last_aux_file_policy.load(),
+            Some(AuxFilePolicy::CrossValidation),
+            "dirty index_part.json reflected state is yet to be updated"
+        );
+
+        // we can still read the auxfile v1 before we ingest anything new
+        let files = tline.list_aux_files(lsn, &ctx).await.unwrap();
+        assert_eq!(
+            files.get("pg_logical/mappings/test1"),
+            Some(&bytes::Bytes::from_static(b"first"))
+        );
+
+        {
+            lsn += 8;
+            let mut modification = tline.begin_modification(lsn);
+            modification
+                .put_file("pg_logical/mappings/test2", b"second", &ctx)
+                .await
+                .unwrap();
+            modification.commit(&ctx).await.unwrap();
+        }
+
+        assert_eq!(
+            tline.last_aux_file_policy.load(),
+            Some(AuxFilePolicy::V2),
+            "ingesting a file should apply the wanted switch state when applicable"
+        );
+
+        let files = tline.list_aux_files(lsn, &ctx).await.unwrap();
+        assert_eq!(
+            files.get("pg_logical/mappings/test1"),
+            Some(&bytes::Bytes::from_static(b"first")),
+            "cross validation writes to both v1 and v2 so this should be available in v2"
+        );
+        assert_eq!(
+            files.get("pg_logical/mappings/test2"),
+            Some(&bytes::Bytes::from_static(b"second"))
+        );
+
+        // mimic again by trying to flip it from V2 to V1 (not switched to while ingesting a file)
+        tenant.set_new_location_config(
+            AttachedTenantConf::try_from(LocationConf::attached_single(
+                TenantConfOpt {
+                    switch_aux_file_policy: Some(AuxFilePolicy::V1),
+                    ..Default::default()
+                },
+                tenant.generation,
+                &pageserver_api::models::ShardParameters::default(),
+            ))
+            .unwrap(),
+        );
+
+        {
+            lsn += 8;
+            let mut modification = tline.begin_modification(lsn);
+            modification
+                .put_file("pg_logical/mappings/test2", b"third", &ctx)
+                .await
+                .unwrap();
+            modification.commit(&ctx).await.unwrap();
+        }
+
+        assert_eq!(
+            tline.get_switch_aux_file_policy(),
+            AuxFilePolicy::V1,
+            "wanted state has been updated again, even if invalid request"
+        );
+
+        assert_eq!(
+            tline.last_aux_file_policy.load(),
+            Some(AuxFilePolicy::V2),
+            "ingesting a file should apply the wanted switch state when applicable"
+        );
+
+        let files = tline.list_aux_files(lsn, &ctx).await.unwrap();
+        assert_eq!(
+            files.get("pg_logical/mappings/test1"),
+            Some(&bytes::Bytes::from_static(b"first"))
+        );
+        assert_eq!(
+            files.get("pg_logical/mappings/test2"),
+            Some(&bytes::Bytes::from_static(b"third"))
+        );
+
+        // mimic again by trying to flip it from from V1 to V2 (not switched to while ingesting a file)
+        tenant.set_new_location_config(
+            AttachedTenantConf::try_from(LocationConf::attached_single(
+                TenantConfOpt {
+                    switch_aux_file_policy: Some(AuxFilePolicy::V2),
+                    ..Default::default()
+                },
+                tenant.generation,
+                &pageserver_api::models::ShardParameters::default(),
+            ))
+            .unwrap(),
+        );
+
+        {
+            lsn += 8;
+            let mut modification = tline.begin_modification(lsn);
+            modification
+                .put_file("pg_logical/mappings/test3", b"last", &ctx)
+                .await
+                .unwrap();
+            modification.commit(&ctx).await.unwrap();
+        }
+
+        assert_eq!(tline.get_switch_aux_file_policy(), AuxFilePolicy::V2);
+
+        assert_eq!(tline.last_aux_file_policy.load(), Some(AuxFilePolicy::V2));
+
+        let files = tline.list_aux_files(lsn, &ctx).await.unwrap();
+        assert_eq!(
+            files.get("pg_logical/mappings/test1"),
+            Some(&bytes::Bytes::from_static(b"first"))
+        );
+        assert_eq!(
+            files.get("pg_logical/mappings/test2"),
+            Some(&bytes::Bytes::from_static(b"third"))
+        );
+        assert_eq!(
+            files.get("pg_logical/mappings/test3"),
+            Some(&bytes::Bytes::from_static(b"last"))
+        );
     }
 }
diff --git a/pageserver/src/tenant/remote_timeline_client/index.rs b/pageserver/src/tenant/remote_timeline_client/index.rs
index 88d4daad5ec8..032dda7ff395 100644
--- a/pageserver/src/tenant/remote_timeline_client/index.rs
+++ b/pageserver/src/tenant/remote_timeline_client/index.rs
@@ -91,10 +91,10 @@ pub struct IndexPart {
     pub(crate) lineage: Lineage,
 
     /// Describes the kind of aux files stored in the timeline.
-    /// 
+    ///
     /// The value is modified during file ingestion when the latest wanted value communicated via tenant config is applied if it is acceptable.
     /// A V1 setting after V2 files have been committed is not accepted.
-    /// 
+    ///
     /// None means no aux files have been written to the storage before the point
     /// when this flag is introduced.
     #[serde(skip_serializing_if = "Option::is_none", default)]

From 61b28ba6ca34988bb0b12e6768a07286ed4f89de Mon Sep 17 00:00:00 2001
From: Alex Chi Z <chi@neon.tech>
Date: Thu, 16 May 2024 14:11:00 -0400
Subject: [PATCH 18/20] add `default_tenant_config`

Signed-off-by: Alex Chi Z <chi@neon.tech>
---
 libs/pageserver_api/src/models.rs   | 5 +++++
 pageserver/src/pgdatadir_mapping.rs | 2 +-
 pageserver/src/tenant/config.rs     | 2 +-
 3 files changed, 7 insertions(+), 2 deletions(-)

diff --git a/libs/pageserver_api/src/models.rs b/libs/pageserver_api/src/models.rs
index ef994436b599..429db15de59d 100644
--- a/libs/pageserver_api/src/models.rs
+++ b/libs/pageserver_api/src/models.rs
@@ -336,6 +336,11 @@ impl AuxFilePolicy {
             (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::V1
+    }
 }
 
 /// The aux file policy memory flag. Users can store `Option<AuxFilePolicy>` into this atomic flag. 0 == unspecified.
diff --git a/pageserver/src/pgdatadir_mapping.rs b/pageserver/src/pgdatadir_mapping.rs
index 23e43e8e9274..402f075365ab 100644
--- a/pageserver/src/pgdatadir_mapping.rs
+++ b/pageserver/src/pgdatadir_mapping.rs
@@ -1487,7 +1487,7 @@ impl<'a> DatadirModification<'a> {
             } else {
                 // This branch handles non-valid migration path, and the case that switch_policy == current_policy.
                 // And actually, because the migration path always allow unspecified -> *, this unwrap_or will never be hit.
-                current_policy.unwrap_or(AuxFilePolicy::V1)
+                current_policy.unwrap_or(AuxFilePolicy::default_tenant_config())
             }
         };
 
diff --git a/pageserver/src/tenant/config.rs b/pageserver/src/tenant/config.rs
index d96766ed05b9..a695363cdced 100644
--- a/pageserver/src/tenant/config.rs
+++ b/pageserver/src/tenant/config.rs
@@ -576,7 +576,7 @@ impl Default for TenantConf {
             lazy_slru_download: false,
             timeline_get_throttle: crate::tenant::throttle::Config::disabled(),
             image_layer_creation_check_threshold: DEFAULT_IMAGE_LAYER_CREATION_CHECK_THRESHOLD,
-            switch_aux_file_policy: AuxFilePolicy::V1,
+            switch_aux_file_policy: AuxFilePolicy::default_tenant_config(),
         }
     }
 }

From 76d4e7712f7ce33fd31ddf0c97cd50d5a8ae8254 Mon Sep 17 00:00:00 2001
From: Alex Chi Z <chi@neon.tech>
Date: Thu, 16 May 2024 16:42:32 -0400
Subject: [PATCH 19/20] nit

Signed-off-by: Alex Chi Z <chi@neon.tech>
---
 pageserver/src/tenant.rs | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/pageserver/src/tenant.rs b/pageserver/src/tenant.rs
index 136b3d3e0f4d..7f44f32c5c95 100644
--- a/pageserver/src/tenant.rs
+++ b/pageserver/src/tenant.rs
@@ -5671,7 +5671,7 @@ mod tests {
             "aux file is written with switch_aux_file_policy unset (which is v1), so we should keep v1"
         );
 
-        // we can still read the auxfile v1 before we ingest anything new
+        // we can read everything from the storage
         let files = tline.list_aux_files(lsn, &ctx).await.unwrap();
         assert_eq!(
             files.get("pg_logical/mappings/test1"),
@@ -5691,7 +5691,7 @@ mod tests {
         assert_eq!(
             tline.last_aux_file_policy.load(),
             Some(AuxFilePolicy::V1),
-            "keep v1 storage format on branching"
+            "keep v1 storage format when new files are written"
         );
 
         let files = tline.list_aux_files(lsn, &ctx).await.unwrap();

From 24a864a10ea591ef00f263fa18269c5964e0a595 Mon Sep 17 00:00:00 2001
From: Alex Chi Z <chi@neon.tech>
Date: Fri, 17 May 2024 11:50:03 -0400
Subject: [PATCH 20/20] rm seqcst comment

Signed-off-by: Alex Chi Z <chi@neon.tech>
---
 libs/pageserver_api/src/models.rs | 2 --
 1 file changed, 2 deletions(-)

diff --git a/libs/pageserver_api/src/models.rs b/libs/pageserver_api/src/models.rs
index 429db15de59d..d612cdedfe52 100644
--- a/libs/pageserver_api/src/models.rs
+++ b/libs/pageserver_api/src/models.rs
@@ -354,8 +354,6 @@ impl AtomicAuxFilePolicy {
     }
 
     pub fn load(&self) -> Option<AuxFilePolicy> {
-        // Using SeqCst memory order because this flag will be read from both write/read threads
-        // and we want to guarantee the strongest consistency.
         match self.0.load(std::sync::atomic::Ordering::Acquire) {
             0 => None,
             other => Some(AuxFilePolicy::from_usize(other)),