Add simple exponential backoff for tenant attach

Cargo.lock

@@ -753,7 +753,6 @@ dependencies = [
  "axum",
  "axum-core",
  "bytes",
- "form_urlencoded",
  "futures-util",
  "headers",
  "http 1.1.0",
@@ -762,8 +761,6 @@ dependencies = [
  "mime",
  "pin-project-lite",
  "serde",
- "serde_html_form",
- "serde_path_to_error",
  "tower 0.5.2",
  "tower-layer",
  "tower-service",
@@ -6425,19 +6422,6 @@ dependencies = [
  "syn 2.0.100",
 ]
 
-[[package]]
-name = "serde_html_form"
-version = "0.2.7"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9d2de91cf02bbc07cde38891769ccd5d4f073d22a40683aa4bc7a95781aaa2c4"
-dependencies = [
- "form_urlencoded",
- "indexmap 2.9.0",
- "itoa",
- "ryu",
- "serde",
-]
-
 [[package]]
 name = "serde_json"
 version = "1.0.125"

Cargo.toml

@@ -71,7 +71,7 @@ aws-credential-types = "1.2.0"
 aws-sigv4 = { version = "1.2", features = ["sign-http"] }
 aws-types = "1.3"
 axum = { version = "0.8.1", features = ["ws"] }
-axum-extra = { version = "0.10.0", features = ["typed-header", "query"] }
+axum-extra = { version = "0.10.0", features = ["typed-header"] }
 base64 = "0.13.0"
 bincode = "1.3"
 bindgen = "0.71"

compute_tools/src/compute.rs

@@ -785,7 +785,7 @@ impl ComputeNode {
         self.spawn_extension_stats_task();
 
         if pspec.spec.autoprewarm {
-            self.prewarm_lfc(None);
+            self.prewarm_lfc();
         }
         Ok(())
     }

compute_tools/src/compute_prewarm.rs

@@ -25,16 +25,11 @@ struct EndpointStoragePair {
 }
 
 const KEY: &str = "lfc_state";
-impl EndpointStoragePair {
-    /// endpoint_id is set to None while prewarming from other endpoint, see replica promotion
-    /// If not None, takes precedence over pspec.spec.endpoint_id
-    fn from_spec_and_endpoint(
-        pspec: &crate::compute::ParsedSpec,
-        endpoint_id: Option<String>,
-    ) -> Result<Self> {
-        let endpoint_id = endpoint_id.as_ref().or(pspec.spec.endpoint_id.as_ref());
-        let Some(ref endpoint_id) = endpoint_id else {
-            bail!("pspec.endpoint_id missing, other endpoint_id not provided")
+impl TryFrom<&crate::compute::ParsedSpec> for EndpointStoragePair {
+    type Error = anyhow::Error;
+    fn try_from(pspec: &crate::compute::ParsedSpec) -> Result<Self, Self::Error> {
+        let Some(ref endpoint_id) = pspec.spec.endpoint_id else {
+            bail!("pspec.endpoint_id missing")
         };
         let Some(ref base_uri) = pspec.endpoint_storage_addr else {
             bail!("pspec.endpoint_storage_addr missing")
@@ -89,7 +84,7 @@ impl ComputeNode {
     }
 
     /// Returns false if there is a prewarm request ongoing, true otherwise
-    pub fn prewarm_lfc(self: &Arc<Self>, from_endpoint: Option<String>) -> bool {
+    pub fn prewarm_lfc(self: &Arc<Self>) -> bool {
         crate::metrics::LFC_PREWARM_REQUESTS.inc();
         {
             let state = &mut self.state.lock().unwrap().lfc_prewarm_state;
@@ -102,7 +97,7 @@ impl ComputeNode {
 
         let cloned = self.clone();
         spawn(async move {
-            let Err(err) = cloned.prewarm_impl(from_endpoint).await else {
+            let Err(err) = cloned.prewarm_impl().await else {
                 cloned.state.lock().unwrap().lfc_prewarm_state = LfcPrewarmState::Completed;
                 return;
             };
@@ -114,14 +109,13 @@ impl ComputeNode {
         true
     }
 
-    /// from_endpoint: None for endpoint managed by this compute_ctl
-    fn endpoint_storage_pair(&self, from_endpoint: Option<String>) -> Result<EndpointStoragePair> {
+    fn endpoint_storage_pair(&self) -> Result<EndpointStoragePair> {
         let state = self.state.lock().unwrap();
-        EndpointStoragePair::from_spec_and_endpoint(state.pspec.as_ref().unwrap(), from_endpoint)
+        state.pspec.as_ref().unwrap().try_into()
     }
 
-    async fn prewarm_impl(&self, from_endpoint: Option<String>) -> Result<()> {
-        let EndpointStoragePair { url, token } = self.endpoint_storage_pair(from_endpoint)?;
+    async fn prewarm_impl(&self) -> Result<()> {
+        let EndpointStoragePair { url, token } = self.endpoint_storage_pair()?;
         info!(%url, "requesting LFC state from endpoint storage");
 
         let request = Client::new().get(&url).bearer_auth(token);
@@ -179,7 +173,7 @@ impl ComputeNode {
     }
 
     async fn offload_lfc_impl(&self) -> Result<()> {
-        let EndpointStoragePair { url, token } = self.endpoint_storage_pair(None)?;
+        let EndpointStoragePair { url, token } = self.endpoint_storage_pair()?;
         info!(%url, "requesting LFC state from postgres");
 
         let mut compressed = Vec::new();

compute_tools/src/http/routes/lfc.rs

@@ -2,7 +2,6 @@ use crate::compute_prewarm::LfcPrewarmStateWithProgress;
 use crate::http::JsonResponse;
 use axum::response::{IntoResponse, Response};
 use axum::{Json, http::StatusCode};
-use axum_extra::extract::OptionalQuery;
 use compute_api::responses::LfcOffloadState;
 type Compute = axum::extract::State<std::sync::Arc<crate::compute::ComputeNode>>;
 
@@ -17,16 +16,8 @@ pub(in crate::http) async fn offload_state(compute: Compute) -> Json<LfcOffloadS
     Json(compute.lfc_offload_state())
 }
 
-#[derive(serde::Deserialize)]
-pub struct PrewarmQuery {
-    pub from_endpoint: String,
-}
-
-pub(in crate::http) async fn prewarm(
-    compute: Compute,
-    OptionalQuery(query): OptionalQuery<PrewarmQuery>,
-) -> Response {
-    if compute.prewarm_lfc(query.map(|q| q.from_endpoint)) {
+pub(in crate::http) async fn prewarm(compute: Compute) -> Response {
+    if compute.prewarm_lfc() {
         StatusCode::ACCEPTED.into_response()
     } else {
         JsonResponse::error(

pageserver/src/tenant.rs

@@ -1535,20 +1535,26 @@ impl TenantShard {
 
                 // We will time the duration of the attach phase unless this is a creation (attach will do no work)
                 let attach_start = std::time::Instant::now();
-                let attached = {
-                    let _attach_timer = Some(TENANT.attach.start_timer());
-                    tenant_clone.attach(preload, &ctx).await
-                };
-                let attach_duration = attach_start.elapsed();
-                _ = tenant_clone.attach_wal_lag_cooldown.set(WalLagCooldown::new(attach_start, attach_duration));
+				_ = backoff::retry(
+					|| async {
+						let attached = {
+							let _attach_timer = Some(TENANT.attach.start_timer());
+							tenant_clone.attach(preload, &ctx).await
+						};
+						let attach_duration = attach_start.elapsed();
+						_ = tenant_clone.attach_wal_lag_cooldown
+							.set(WalLagCooldown::new(attach_start, attach_duration));
+						attached
+					},
+					|_| false,
+					3,
+					u32::MAX,
+					"Retry any S3 operations.",
+					&tenant_clone.cancel
+				);
 
-                match attached {
-                    Ok(()) => {
-                        info!("attach finished, activating");
-                        tenant_clone.activate(broker_client, None, &ctx);
-                    }
-                    Err(e) => make_broken_or_stopping(&tenant_clone, anyhow::anyhow!(e)),
-                }
+                info!("attach finished, activating");
+                tenant_clone.activate(broker_client, None, &ctx);
 
                 // If we are doing an opportunistic warmup attachment at startup, initialize
                 // logical size at the same time.  This is better than starting a bunch of idle tenants

pgxn/neon/neon_lwlsncache.c

@@ -11,9 +11,6 @@
 #include "utils/guc.h"
 #include "utils/hsearch.h"
 
-#if PG_MAJORVERSION_NUM > 14
-#include "access/xlogrecovery.h"
-#endif
 
 
 typedef struct LastWrittenLsnCacheEntry
@@ -27,20 +24,14 @@ typedef struct LastWrittenLsnCacheEntry
 typedef struct LwLsnCacheCtl {
 	int lastWrittenLsnCacheSize;
 	/*
-	 * Highest (most recent) last written LSN, for pages not present in
-	 * lastWrittenLsnCache
-	 */
-	XLogRecPtr  maxLastWrittenLsnData;
+	* Maximal last written LSN for pages not present in lastWrittenLsnCache
+	*/
+	XLogRecPtr  maxLastWrittenLsn;
 
 	/*
-	 * Maximal last written LSN for metadata, not present in lastWrittenLsnCache
-	 */
-	XLogRecPtr  maxLastWrittenLsnMetadata;
-
-	/*
-	 * Double linked list to implement LRU replacement policy for last written LSN cache.
-	 * Access to this list as well as to last written LSN cache is protected by 'LastWrittenLsnLock'.
-	 */
+	* Double linked list to implement LRU replacement policy for last written LSN cache.
+	* Access to this list as well as to last written LSN cache is protected by 'LastWrittenLsnLock'.
+	*/
 	dlist_head lastWrittenLsnLRU;
 } LwLsnCacheCtl;
 
@@ -117,20 +108,19 @@ init_lwlsncache(void)
 	#else
 	shmemrequest();
 	#endif
-
-#define SET_HOOK(name) do { \
-	prev_##name##_hook = name##_hook; \
-	name##_hook = neon_##name; \
-} while (false)
-
-	SET_HOOK(set_lwlsn_block_range);
-	SET_HOOK(set_lwlsn_block_v);
-	SET_HOOK(set_lwlsn_block);
-	SET_HOOK(set_max_lwlsn);
-	SET_HOOK(set_lwlsn_relation);
-	SET_HOOK(set_lwlsn_db);
-
-#undef SET_HOOK
+	
+	prev_set_lwlsn_block_range_hook = set_lwlsn_block_range_hook;
+	set_lwlsn_block_range_hook = neon_set_lwlsn_block_range;
+	prev_set_lwlsn_block_v_hook = set_lwlsn_block_v_hook;
+	set_lwlsn_block_v_hook = neon_set_lwlsn_block_v;
+	prev_set_lwlsn_block_hook = set_lwlsn_block_hook;
+	set_lwlsn_block_hook = neon_set_lwlsn_block;
+	prev_set_max_lwlsn_hook = set_max_lwlsn_hook;
+	set_max_lwlsn_hook = neon_set_max_lwlsn;
+	prev_set_lwlsn_relation_hook = set_lwlsn_relation_hook;
+	set_lwlsn_relation_hook = neon_set_lwlsn_relation;
+	prev_set_lwlsn_db_hook = set_lwlsn_db_hook;
+	set_lwlsn_db_hook = neon_set_lwlsn_db;
 }
 
 
@@ -149,34 +139,24 @@ static void shmemrequest(void) {
 
 static void shmeminit(void) {
 	static HASHCTL info;
-	bool found = true;
-
+	bool found;
 	if (lwlsn_cache_size > 0)
 	{
 		info.keysize = sizeof(BufferTag);
 		info.entrysize = sizeof(LastWrittenLsnCacheEntry);
-
 		lastWrittenLsnCache = ShmemInitHash("last_written_lsn_cache",
-											lwlsn_cache_size, lwlsn_cache_size,
-											&info,
-											HASH_ELEM | HASH_BLOBS);
-		LwLsnCache = ShmemInitStruct("neon/LwLsnCacheCtl",
-									 sizeof(LwLsnCacheCtl), &found);
-	}
-
-	/* initialize the shmem struct if we allocated it */
-	if (!found) {
-		XLogRecPtr redoPtr;
+			lwlsn_cache_size, lwlsn_cache_size,
+										&info,
+										HASH_ELEM | HASH_BLOBS);
+		LwLsnCache = ShmemInitStruct("neon/LwLsnCacheCtl", sizeof(LwLsnCacheCtl), &found);
+		// Now set the size in the struct
 		LwLsnCache->lastWrittenLsnCacheSize = lwlsn_cache_size;
-
-		dlist_init(&LwLsnCache->lastWrittenLsnLRU);
-
-		redoPtr = GetRedoRecPtr();
-
-		LwLsnCache->maxLastWrittenLsnMetadata = redoPtr;
-		LwLsnCache->maxLastWrittenLsnData = redoPtr;
+		if (found) {
+			return;
+		}
 	}
-
+	dlist_init(&LwLsnCache->lastWrittenLsnLRU);
+    LwLsnCache->maxLastWrittenLsn = GetRedoRecPtr();
 	if (prev_shmem_startup_hook) {
 		prev_shmem_startup_hook();
 	}
@@ -200,18 +180,17 @@ neon_get_lwlsn(NRelFileInfo rlocator, ForkNumber forknum, BlockNumber blkno)
 
 	LWLockAcquire(LastWrittenLsnLock, LW_SHARED);
 
-	if (NInfoGetRelNumber(rlocator) != InvalidOid) /* data page*/
+	/* Maximal last written LSN among all non-cached pages */
+	lsn = LwLsnCache->maxLastWrittenLsn;
+
+	if (NInfoGetRelNumber(rlocator) != InvalidOid)
 	{
 		BufferTag key;
 		Oid spcOid = NInfoGetSpcOid(rlocator);
 		Oid dbOid = NInfoGetDbOid(rlocator);
 		Oid relNumber = NInfoGetRelNumber(rlocator);
-
 		BufTagInit(key,  relNumber, forknum, blkno, spcOid, dbOid);
-
-		/* Maximal last written LSN among all non-cached data pages */
-		lsn = LwLsnCache->maxLastWrittenLsnData;
-
+		
 		entry = hash_search(lastWrittenLsnCache, &key, HASH_FIND, NULL);
 		if (entry != NULL)
 			lsn = entry->lsn;
@@ -233,13 +212,9 @@ neon_get_lwlsn(NRelFileInfo rlocator, ForkNumber forknum, BlockNumber blkno)
 			 lsn = SetLastWrittenLSNForBlockRangeInternal(lsn, rlocator, forknum, blkno, 1);
 		}
 	}
-	else /* metadata */
+	else
 	{
 		HASH_SEQ_STATUS seq;
-		/* Maximal last written LSN for metadata */
-		lsn = Max(LwLsnCache->maxLastWrittenLsnMetadata,
-				  LwLsnCache->maxLastWrittenLsnData);
-
 		/* Find maximum of all cached LSNs */
 		hash_seq_init(&seq, lastWrittenLsnCache);
 		while ((entry = (LastWrittenLsnCacheEntry *) hash_seq_search(&seq)) != NULL)
@@ -255,8 +230,7 @@ neon_get_lwlsn(NRelFileInfo rlocator, ForkNumber forknum, BlockNumber blkno)
 
 static void neon_set_max_lwlsn(XLogRecPtr lsn) {
 	LWLockAcquire(LastWrittenLsnLock, LW_EXCLUSIVE);
-	LwLsnCache->maxLastWrittenLsnMetadata = lsn;
-	LwLsnCache->maxLastWrittenLsnData = lsn;
+	LwLsnCache->maxLastWrittenLsn = lsn;
 	LWLockRelease(LastWrittenLsnLock);
 }
 
@@ -317,7 +291,7 @@ neon_get_lwlsn_v(NRelFileInfo relfilenode, ForkNumber forknum,
 			LWLockRelease(LastWrittenLsnLock);
 			LWLockAcquire(LastWrittenLsnLock, LW_EXCLUSIVE);
 
-			lsn = LwLsnCache->maxLastWrittenLsnData;
+			lsn = LwLsnCache->maxLastWrittenLsn;
 
 			for (int i = 0; i < nblocks; i++)
 			{
@@ -332,8 +306,7 @@ neon_get_lwlsn_v(NRelFileInfo relfilenode, ForkNumber forknum,
 	else
 	{
 		HASH_SEQ_STATUS seq;
-		Assert(nblocks == 1);
-		lsn = LwLsnCache->maxLastWrittenLsnMetadata;
+		lsn = LwLsnCache->maxLastWrittenLsn;
 		/* Find maximum of all cached LSNs */
 		hash_seq_init(&seq, lastWrittenLsnCache);
 		while ((entry = (LastWrittenLsnCacheEntry *) hash_seq_search(&seq)) != NULL)
@@ -361,10 +334,10 @@ SetLastWrittenLSNForBlockRangeInternal(XLogRecPtr lsn,
 {
 	if (NInfoGetRelNumber(rlocator) == InvalidOid)
 	{
-		if (lsn > LwLsnCache->maxLastWrittenLsnMetadata)
-			LwLsnCache->maxLastWrittenLsnMetadata = lsn;
+		if (lsn > LwLsnCache->maxLastWrittenLsn)
+		LwLsnCache->maxLastWrittenLsn = lsn;
 		else
-			lsn = LwLsnCache->maxLastWrittenLsnMetadata;
+			lsn = LwLsnCache->maxLastWrittenLsn;
 	}
 	else
 	{
@@ -396,19 +369,10 @@ SetLastWrittenLSNForBlockRangeInternal(XLogRecPtr lsn,
 				if (hash_get_num_entries(lastWrittenLsnCache) > LwLsnCache->lastWrittenLsnCacheSize)
 				{
 					/* Replace least recently used entry */
-					LastWrittenLsnCacheEntry* victim = NULL;
-					victim = dlist_container(LastWrittenLsnCacheEntry, lru_node, dlist_pop_head_node(&LwLsnCache->lastWrittenLsnLRU));
-
-					while (!XLogRecordReplayFinished(victim->lsn))
-					{
-						/* in recovery, we don't allow eviction of entries with the LSN of a record that has yet to be returned */
-						dlist_push_tail(&LwLsnCache->lastWrittenLsnLRU, &entry->lru_node);
-						victim = dlist_container(LastWrittenLsnCacheEntry, lru_node, dlist_pop_head_node(&LwLsnCache->lastWrittenLsnLRU));
-					}
-
+					LastWrittenLsnCacheEntry* victim = dlist_container(LastWrittenLsnCacheEntry, lru_node, dlist_pop_head_node(&LwLsnCache->lastWrittenLsnLRU));
 					/* Adjust max LSN for not cached relations/chunks if needed */
-					if (victim->lsn > LwLsnCache->maxLastWrittenLsnMetadata)
-						LwLsnCache->maxLastWrittenLsnMetadata = victim->lsn;
+					if (victim->lsn > LwLsnCache->maxLastWrittenLsn)
+					LwLsnCache->maxLastWrittenLsn = victim->lsn;
 
 					hash_search(lastWrittenLsnCache, victim, HASH_REMOVE, NULL);
 				}
@@ -469,13 +433,6 @@ neon_set_lwlsn_block_v(const XLogRecPtr *lsns, NRelFileInfo relfilenode,
 	Oid dbOid = NInfoGetDbOid(relfilenode);
 	Oid relNumber = NInfoGetRelNumber(relfilenode);
 
-	/*
-	 * We ignore the operation when the input is invalid:
-	 *  - we must have gotten LSNs to set
-	 *  - we must have pages to write
-	 *  - the cache must be enabled
-	 *  - we must be processing a data page, not a metadata request
-	 */
 	if (lsns == NULL || nblocks == 0 || LwLsnCache->lastWrittenLsnCacheSize == 0 ||
 		NInfoGetRelNumber(relfilenode) == InvalidOid)
 		return InvalidXLogRecPtr;
@@ -509,25 +466,10 @@ neon_set_lwlsn_block_v(const XLogRecPtr *lsns, NRelFileInfo relfilenode,
 			if (hash_get_num_entries(lastWrittenLsnCache) > LwLsnCache->lastWrittenLsnCacheSize)
 			{
 				/* Replace least recently used entry */
-				LastWrittenLsnCacheEntry* victim = dlist_container(LastWrittenLsnCacheEntry, lru_node,
-																   dlist_pop_head_node(&LwLsnCache->lastWrittenLsnLRU));
-
-				/*
-				 * If replay is still working on this LSN, we can't evict the
-				 * page. Therefore, we must find a different victim, and return
-				 * the one we just found to the pool.
-				 */
-				while (!XLogRecordReplayFinished(victim->lsn))
-				{
-					dlist_push_tail(&LwLsnCache->lastWrittenLsnLRU,
-									&entry->lru_node);
-					victim = dlist_container(LastWrittenLsnCacheEntry, lru_node,
-											 dlist_pop_head_node(&LwLsnCache->lastWrittenLsnLRU));
-				}
-
+				LastWrittenLsnCacheEntry* victim = dlist_container(LastWrittenLsnCacheEntry, lru_node, dlist_pop_head_node(&LwLsnCache->lastWrittenLsnLRU));
 				/* Adjust max LSN for not cached relations/chunks if needed */
-				if (victim->lsn > LwLsnCache->maxLastWrittenLsnData)
-					LwLsnCache->maxLastWrittenLsnData = victim->lsn;
+				if (victim->lsn > LwLsnCache->maxLastWrittenLsn)
+					LwLsnCache->maxLastWrittenLsn = victim->lsn;
 
 				hash_search(lastWrittenLsnCache, victim, HASH_REMOVE, NULL);
 			}

test_runner/fixtures/endpoint/http.py

@@ -69,10 +69,8 @@ class EndpointHttpClient(requests.Session):
         json: dict[str, str] = res.json()
         return json
 
-    def prewarm_lfc(self, from_endpoint_id: str | None = None):
-        url: str = f"http://localhost:{self.external_port}/lfc/prewarm"
-        params = {"from_endpoint": from_endpoint_id} if from_endpoint_id else dict()
-        self.post(url, params=params).raise_for_status()
+    def prewarm_lfc(self):
+        self.post(f"http://localhost:{self.external_port}/lfc/prewarm").raise_for_status()
 
         def prewarmed():
             json = self.prewarm_lfc_status()

test_runner/regress/test_lfc_prewarm.py

@@ -188,8 +188,7 @@ def test_lfc_prewarm_under_workload(neon_simple_env: NeonEnv, query: LfcQueryMet
             pg_cur.execute("select pg_reload_conf()")
 
             if query is LfcQueryMethod.COMPUTE_CTL:
-                # Same thing as prewarm_lfc(), testing other method
-                http_client.prewarm_lfc(endpoint.endpoint_id)
+                http_client.prewarm_lfc()
             else:
                 pg_cur.execute("select prewarm_local_cache(%s)", (lfc_state,))