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/communicator.c

@@ -2085,6 +2085,9 @@ communicator_read_at_lsnv(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber ba
 
 		start_ts = GetCurrentTimestamp();
 
+		if (RecoveryInProgress() && MyBackendType != B_STARTUP)
+			XLogWaitForReplayOf(reqlsns->request_lsn);
+
 		/*
 		 * Try to find prefetched page in the list of received pages.
 		 */

pgxn/neon/neon--1.6--1.5.sql

@@ -2,6 +2,6 @@ DROP FUNCTION IF EXISTS get_prewarm_info(out total_pages integer, out prewarmed_
 
 DROP FUNCTION IF EXISTS get_local_cache_state(max_chunks integer);
 
-DROP FUNCTION IF EXISTS prewarm_local_cache(state bytea, n_workers integer);
+DROP FUNCTION IF EXISTS prewarm_local_cache(state bytea, n_workers integer default 1);
 
 

pgxn/neon/pagestore_smgr.c

@@ -2382,8 +2382,6 @@ get_fsm_physical_block(BlockNumber heapblk)
 static bool
 neon_redo_read_buffer_filter(XLogReaderState *record, uint8 block_id)
 {
-	static XLogRecPtr last_recptr = InvalidXLogRecPtr;
-	static bool last_no_redo_needed;
 	XLogRecPtr	end_recptr = record->EndRecPtr;
 	NRelFileInfo rinfo;
 	ForkNumber	forknum;
@@ -2392,88 +2390,69 @@ neon_redo_read_buffer_filter(XLogReaderState *record, uint8 block_id)
 	uint32		hash;
 	LWLock	   *partitionLock;
 	int			buf_id;
-	bool		no_redo_needed = true;
+	bool		no_redo_needed;
 
 	if (old_redo_read_buffer_filter && old_redo_read_buffer_filter(record, block_id))
 		return true;
 
-	if (last_recptr != end_recptr)
+#if PG_VERSION_NUM < 150000
+	if (!XLogRecGetBlockTag(record, block_id, &rinfo, &forknum, &blkno))
+		neon_log(PANIC, "failed to locate backup block with ID %d", block_id);
+#else
+	XLogRecGetBlockTag(record, block_id, &rinfo, &forknum, &blkno);
+#endif
+
+	CopyNRelFileInfoToBufTag(tag, rinfo);
+	tag.forkNum = forknum;
+	tag.blockNum = blkno;
+
+	hash = BufTableHashCode(&tag);
+	partitionLock = BufMappingPartitionLock(hash);
+
+	/*
+	 * Lock the partition of shared_buffers so that it can't be updated
+	 * concurrently.
+	 */
+	LWLockAcquire(partitionLock, LW_SHARED);
+
+	/*
+	 * Out of an abundance of caution, we always run redo on shared catalogs,
+	 * regardless of whether the block is stored in shared buffers. See also
+	 * this function's top comment.
+	 */
+	if (!OidIsValid(NInfoGetDbOid(rinfo)))
 	{
-#if PG_VERSION_NUM < 150000
-		int max_block_id = record->max_block_id;
-#else
-		int max_block_id = XLogRecMaxBlockId(record);
-#endif
-		for (int block_id = 0; block_id <= max_block_id && no_redo_needed; block_id++)
-		{
-			if (XLogRecHasBlockRef(record, block_id))
-			{
-#if PG_VERSION_NUM < 150000
-				if (!XLogRecGetBlockTag(record, block_id, &rinfo, &forknum, &blkno))
-					neon_log(PANIC, "failed to locate backup block with ID %d", block_id);
-#else
-				XLogRecGetBlockTag(record, block_id, &rinfo, &forknum, &blkno);
-#endif
-				CopyNRelFileInfoToBufTag(tag, rinfo);
-				tag.forkNum = forknum;
-				tag.blockNum = blkno;
-
-				hash = BufTableHashCode(&tag);
-				partitionLock = BufMappingPartitionLock(hash);
-
-				/*
-				 * Lock the partition of shared_buffers so that it can't be updated
-				 * concurrently.
-				 */
-				LWLockAcquire(partitionLock, LW_SHARED);
-
-				/*
-				 * Out of an abundance of caution, we always run redo on shared catalogs,
-				 * regardless of whether the block is stored in shared buffers. See also
-				 * this function's top comment.
-				 */
-				if (!OidIsValid(NInfoGetDbOid(rinfo)))
-				{
-					no_redo_needed = false;
-				}
-				else
-				{
-					/* Try to find the relevant buffer */
-					buf_id = BufTableLookup(&tag, hash);
-
-					no_redo_needed = buf_id < 0;
-				}
-
-				/*
-				 * we don't have the buffer in memory, update lwLsn past this record, also
-				 * evict page from file cache
-				 */
-				if (no_redo_needed)
-				{
-					neon_set_lwlsn_block(end_recptr, rinfo, forknum, blkno);
-					/*
-					 * Redo changes if page exists in LFC.
-					 * We should perform this check after assigning LwLSN to prevent
-					 * prefetching of some older version of the page by some other backend.
-					 */
-					no_redo_needed = !lfc_cache_contains(rinfo, forknum, blkno);
-				}
-
-				LWLockRelease(partitionLock);
-
-				neon_extend_rel_size(rinfo, forknum, blkno, end_recptr);
-				if (forknum == MAIN_FORKNUM)
-				{
-					neon_extend_rel_size(rinfo, FSM_FORKNUM, get_fsm_physical_block(blkno), end_recptr);
-				}
-			}
-		}
-		last_recptr = end_recptr;
-		last_no_redo_needed = no_redo_needed;
+		no_redo_needed = false;
 	}
 	else
 	{
-		no_redo_needed = last_no_redo_needed;
+		/* Try to find the relevant buffer */
+		buf_id = BufTableLookup(&tag, hash);
+
+		no_redo_needed = buf_id < 0;
+	}
+
+	/*
+	 * we don't have the buffer in memory, update lwLsn past this record, also
+	 * evict page from file cache
+	 */
+	if (no_redo_needed)
+	{
+		neon_set_lwlsn_block(end_recptr, rinfo, forknum, blkno);
+		/*
+		 * Redo changes if page exists in LFC.
+		 * We should perform this check after assigning LwLSN to prevent
+		 * prefetching of some older version of the page by some other backend.
+		 */
+		no_redo_needed = !lfc_cache_contains(rinfo, forknum, blkno);
+	}
+
+	LWLockRelease(partitionLock);
+
+	neon_extend_rel_size(rinfo, forknum, blkno, end_recptr);
+	if (forknum == MAIN_FORKNUM)
+	{
+		neon_extend_rel_size(rinfo, FSM_FORKNUM, get_fsm_physical_block(blkno), end_recptr);
 	}
 	return no_redo_needed;
 }

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,))