feat(pageserver): support multiple key ranges for image initial flush path

Signed-off-by: Alex Chi Z <chi@neon.tech>

Dockerfile.compute-node

@@ -243,12 +243,15 @@ COPY --from=pg-build /usr/local/pgsql/ /usr/local/pgsql/
 
 COPY patches/pgvector.patch /pgvector.patch
 
+# By default, pgvector Makefile uses `-march=native`. We don't want that, 
+# because we build the images on different machines than where we run them.
+# Pass OPTFLAGS="" to remove it.
 RUN wget https://github.com/pgvector/pgvector/archive/refs/tags/v0.7.0.tar.gz -O pgvector.tar.gz && \
     echo "1b5503a35c265408b6eb282621c5e1e75f7801afc04eecb950796cfee2e3d1d8 pgvector.tar.gz" | sha256sum --check && \
     mkdir pgvector-src && cd pgvector-src && tar xvzf ../pgvector.tar.gz --strip-components=1 -C . && \
     patch -p1 < /pgvector.patch && \
-    make -j $(getconf _NPROCESSORS_ONLN) PG_CONFIG=/usr/local/pgsql/bin/pg_config && \
-    make -j $(getconf _NPROCESSORS_ONLN) install PG_CONFIG=/usr/local/pgsql/bin/pg_config && \
+    make -j $(getconf _NPROCESSORS_ONLN) OPTFLAGS="" PG_CONFIG=/usr/local/pgsql/bin/pg_config && \
+    make -j $(getconf _NPROCESSORS_ONLN) OPTFLAGS="" install PG_CONFIG=/usr/local/pgsql/bin/pg_config && \
     echo 'trusted = true' >> /usr/local/pgsql/share/extension/vector.control
 
 #########################################################################################

libs/postgres_connection/src/lib.rs

@@ -178,6 +178,13 @@ impl PgConnectionConfig {
     }
 }
 
+impl fmt::Display for PgConnectionConfig {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        // The password is intentionally hidden and not part of this display string.
+        write!(f, "postgresql://{}:{}", self.host, self.port)
+    }
+}
+
 impl fmt::Debug for PgConnectionConfig {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         // We want `password: Some(REDACTED-STRING)`, not `password: Some("REDACTED-STRING")`

pageserver/src/http/routes.rs

@@ -395,7 +395,7 @@ async fn build_timeline_info_common(
         let guard = timeline.last_received_wal.lock().unwrap();
         if let Some(info) = guard.as_ref() {
             (
-                Some(format!("{:?}", info.wal_source_connconf)), // Password is hidden, but it's for statistics only.
+                Some(format!("{}", info.wal_source_connconf)), // Password is hidden, but it's for statistics only.
                 Some(info.last_received_msg_lsn),
                 Some(info.last_received_msg_ts),
             )

pageserver/src/tenant/blob_io.rs

@@ -238,10 +238,13 @@ impl<const BUFFERED: bool> BlobWriter<BUFFERED> {
                         io_buf,
                         Err(Error::new(
                             ErrorKind::Other,
-                            format!("blob too large ({} bytes)", len),
+                            format!("blob too large ({len} bytes)"),
                         )),
                     );
                 }
+                if len > 0x0fff_ffff {
+                    tracing::warn!("writing blob above future limit ({len} bytes)");
+                }
                 let mut len_buf = (len as u32).to_be_bytes();
                 len_buf[0] |= 0x80;
                 io_buf.extend_from_slice(&len_buf[..]);

pageserver/src/tenant/secondary/downloader.rs

@@ -93,7 +93,7 @@ pub(super) async fn downloader_task(
 
     scheduler
         .run(command_queue, background_jobs_can_start, cancel)
-        .instrument(info_span!("secondary_downloads"))
+        .instrument(info_span!("secondary_download_scheduler"))
         .await
 }
 
@@ -1013,6 +1013,11 @@ impl<'a> TenantDownloader<'a> {
         );
 
         // Note: no backoff::retry wrapper here because download_layer_file does its own retries internally
+        tracing::info!(
+            "Starting download of layer {}, size {}",
+            layer.name,
+            layer.metadata.file_size
+        );
         let downloaded_bytes = match download_layer_file(
             self.conf,
             self.remote_storage,

pageserver/src/tenant/secondary/heatmap_uploader.rs

@@ -53,7 +53,7 @@ pub(super) async fn heatmap_uploader_task(
 
     scheduler
         .run(command_queue, background_jobs_can_start, cancel)
-        .instrument(info_span!("heatmap_uploader"))
+        .instrument(info_span!("heatmap_upload_scheduler"))
         .await
 }
 

pageserver/src/tenant/secondary/scheduler.rs

@@ -179,6 +179,13 @@ where
             // Schedule some work, if concurrency limit permits it
             self.spawn_pending();
 
+            // This message is printed every scheduling iteration as proof of liveness when looking at logs
+            tracing::info!(
+                "Status: {} tasks running, {} pending",
+                self.running.len(),
+                self.pending.len()
+            );
+
             // Between scheduling iterations, we will:
             //  - Drain any complete tasks and spawn pending tasks
             //  - Handle incoming administrative commands
@@ -258,7 +265,11 @@ where
 
         self.tasks.spawn(fut);
 
-        self.running.insert(tenant_shard_id, in_progress);
+        let replaced = self.running.insert(tenant_shard_id, in_progress);
+        debug_assert!(replaced.is_none());
+        if replaced.is_some() {
+            tracing::warn!(%tenant_shard_id, "Unexpectedly spawned a task when one was already running")
+        }
     }
 
     /// For all pending tenants that are elegible for execution, spawn their task.
@@ -268,7 +279,9 @@ where
         while !self.pending.is_empty() && self.running.len() < self.concurrency {
             // unwrap: loop condition includes !is_empty()
             let pending = self.pending.pop_front().unwrap();
-            self.do_spawn(pending);
+            if !self.running.contains_key(pending.get_tenant_shard_id()) {
+                self.do_spawn(pending);
+            }
         }
     }
 
@@ -321,7 +334,8 @@ where
 
         let tenant_shard_id = job.get_tenant_shard_id();
         let barrier = if let Some(barrier) = self.get_running(tenant_shard_id) {
-            tracing::info!("Command already running, waiting for it");
+            tracing::info!(tenant_id=%tenant_shard_id.tenant_id, shard_id=%tenant_shard_id.shard_slug(),
+                           "Command already running, waiting for it");
             barrier
         } else {
             let running = self.spawn_now(job);

pageserver/src/tenant/storage_layer/image_layer.rs

@@ -650,7 +650,7 @@ impl ImageLayerWriterInner {
                 lsn,
             },
         );
-        info!("new image layer {path}");
+        trace!("creating image layer {}", path);
         let mut file = {
             VirtualFile::open_with_options(
                 &path,
@@ -770,7 +770,7 @@ impl ImageLayerWriterInner {
         // FIXME: why not carry the virtualfile here, it supports renaming?
         let layer = Layer::finish_creating(self.conf, timeline, desc, &self.path)?;
 
-        trace!("created image layer {}", layer.local_path());
+        info!("created image layer {}", layer.local_path());
 
         Ok(layer)
     }

pageserver/src/tenant/timeline.rs

@@ -3882,7 +3882,7 @@ impl Timeline {
             }
         }
 
-        let (layers_to_upload, delta_layer_to_add) = if create_image_layer {
+        let (layers_to_upload, delta_layers_to_add) = if create_image_layer {
             // Note: The 'ctx' in use here has DownloadBehavior::Error. We should not
             // require downloading anything during initial import.
             let ((rel_partition, metadata_partition), _lsn) = self
@@ -3899,26 +3899,23 @@ impl Timeline {
             }
 
             // For metadata, always create delta layers.
-            let delta_layer = if !metadata_partition.parts.is_empty() {
-                assert_eq!(
-                    metadata_partition.parts.len(),
-                    1,
-                    "currently sparse keyspace should only contain a single aux file keyspace"
-                );
-                let metadata_keyspace = &metadata_partition.parts[0];
-                assert_eq!(
-                    metadata_keyspace.0.ranges.len(),
-                    1,
-                    "aux file keyspace should be a single range"
-                );
-                self.create_delta_layer(
-                    &frozen_layer,
-                    Some(metadata_keyspace.0.ranges[0].clone()),
-                    ctx,
-                )
-                .await?
+            let delta_layers = if !metadata_partition.parts.is_empty() {
+                // In the current implementation, the metadata partition will only have one part, and the part will only have
+                // one single key range. This might change in the future.
+                let mut delta_layers_created = Vec::new();
+                for ks in &metadata_partition.parts {
+                    for range in &ks.0.ranges {
+                        let layer = self
+                            .create_delta_layer(&frozen_layer, Some(range.clone()), ctx)
+                            .await?;
+                        if let Some(layer) = layer {
+                            delta_layers_created.push(layer);
+                        }
+                    }
+                }
+                delta_layers_created
             } else {
-                None
+                Vec::new()
             };
 
             // For image layers, we add them immediately into the layer map.
@@ -3933,12 +3930,8 @@ impl Timeline {
                 .await?,
             );
 
-            if let Some(delta_layer) = delta_layer {
-                layers_to_upload.push(delta_layer.clone());
-                (layers_to_upload, Some(delta_layer))
-            } else {
-                (layers_to_upload, None)
-            }
+            layers_to_upload.extend(delta_layers.iter().cloned());
+            (layers_to_upload, delta_layers)
         } else {
             // Normal case, write out a L0 delta layer file.
             // `create_delta_layer` will not modify the layer map.
@@ -3946,12 +3939,7 @@ impl Timeline {
             let Some(layer) = self.create_delta_layer(&frozen_layer, None, ctx).await? else {
                 panic!("delta layer cannot be empty if no filter is applied");
             };
-            (
-                // FIXME: even though we have a single image and single delta layer assumption
-                // we push them to vec
-                vec![layer.clone()],
-                Some(layer),
-            )
+            (vec![layer.clone()], vec![layer])
         };
 
         pausable_failpoint!("flush-layer-cancel-after-writing-layer-out-pausable");
@@ -3972,7 +3960,7 @@ impl Timeline {
                 return Err(FlushLayerError::Cancelled);
             }
 
-            guard.finish_flush_l0_layer(delta_layer_to_add.as_ref(), &frozen_layer, &self.metrics);
+            guard.finish_flush_l0_layer(&delta_layers_to_add, &frozen_layer, &self.metrics);
 
             if self.set_disk_consistent_lsn(disk_consistent_lsn) {
                 // Schedule remote uploads that will reflect our new disk_consistent_lsn
@@ -4347,7 +4335,7 @@ impl Timeline {
         let delta_file_accessed = reconstruct_state.get_delta_layers_visited();
 
         let trigger_generation = delta_file_accessed as usize >= MAX_AUX_FILE_V2_DELTAS;
-        info!(
+        debug!(
             "generate image layers for metadata keys: trigger_generation={trigger_generation}, \
                 delta_file_accessed={delta_file_accessed}, total_kb_retrieved={total_kb_retrieved}, \
                 total_key_retrieved={total_key_retrieved}"

pageserver/src/tenant/timeline/compaction.rs

@@ -9,7 +9,10 @@ use std::ops::{Deref, Range};
 use std::sync::Arc;
 
 use super::layer_manager::LayerManager;
-use super::{CompactFlags, DurationRecorder, ImageLayerCreationMode, RecordedDuration, Timeline};
+use super::{
+    CompactFlags, CreateImageLayersError, DurationRecorder, ImageLayerCreationMode,
+    RecordedDuration, Timeline,
+};
 
 use anyhow::{anyhow, Context};
 use enumset::EnumSet;
@@ -22,14 +25,13 @@ use tracing::{debug, info, info_span, trace, warn, Instrument};
 use utils::id::TimelineId;
 
 use crate::context::{AccessStatsBehavior, RequestContext, RequestContextBuilder};
+use crate::page_cache;
 use crate::tenant::storage_layer::{AsLayerDesc, PersistentLayerDesc};
-use crate::tenant::timeline::{drop_rlock, is_rel_fsm_block_key, is_rel_vm_block_key, Hole};
+use crate::tenant::timeline::{drop_rlock, Hole, ImageLayerCreationOutcome};
 use crate::tenant::timeline::{DeltaLayerWriter, ImageLayerWriter};
 use crate::tenant::timeline::{Layer, ResidentLayer};
 use crate::tenant::DeltaLayer;
-use crate::tenant::PageReconstructError;
 use crate::virtual_file::{MaybeFatalIo, VirtualFile};
-use crate::{page_cache, ZERO_PAGE};
 
 use crate::keyspace::KeySpace;
 use crate::repository::Key;
@@ -1150,10 +1152,10 @@ impl TimelineAdaptor {
         lsn: Lsn,
         key_range: &Range<Key>,
         ctx: &RequestContext,
-    ) -> Result<(), PageReconstructError> {
+    ) -> Result<(), CreateImageLayersError> {
         let timer = self.timeline.metrics.create_images_time_histo.start_timer();
 
-        let mut image_layer_writer = ImageLayerWriter::new(
+        let image_layer_writer = ImageLayerWriter::new(
             self.timeline.conf,
             self.timeline.timeline_id,
             self.timeline.tenant_shard_id,
@@ -1164,47 +1166,34 @@ impl TimelineAdaptor {
         .await?;
 
         fail_point!("image-layer-writer-fail-before-finish", |_| {
-            Err(PageReconstructError::Other(anyhow::anyhow!(
+            Err(CreateImageLayersError::Other(anyhow::anyhow!(
                 "failpoint image-layer-writer-fail-before-finish"
             )))
         });
-        let keyspace_ranges = self.get_keyspace(key_range, lsn, ctx).await?;
-        for range in &keyspace_ranges {
-            let mut key = range.start;
-            while key < range.end {
-                let img = match self.timeline.get(key, lsn, ctx).await {
-                    Ok(img) => img,
-                    Err(err) => {
-                        // If we fail to reconstruct a VM or FSM page, we can zero the
-                        // page without losing any actual user data. That seems better
-                        // than failing repeatedly and getting stuck.
-                        //
-                        // We had a bug at one point, where we truncated the FSM and VM
-                        // in the pageserver, but the Postgres didn't know about that
-                        // and continued to generate incremental WAL records for pages
-                        // that didn't exist in the pageserver. Trying to replay those
-                        // WAL records failed to find the previous image of the page.
-                        // This special case allows us to recover from that situation.
-                        // See https://github.com/neondatabase/neon/issues/2601.
-                        //
-                        // Unfortunately we cannot do this for the main fork, or for
-                        // any metadata keys, keys, as that would lead to actual data
-                        // loss.
-                        if is_rel_fsm_block_key(key) || is_rel_vm_block_key(key) {
-                            warn!("could not reconstruct FSM or VM key {key}, filling with zeros: {err:?}");
-                            ZERO_PAGE.clone()
-                        } else {
-                            return Err(err);
-                        }
-                    }
-                };
-                image_layer_writer.put_image(key, img, ctx).await?;
-                key = key.next();
-            }
-        }
-        let image_layer = image_layer_writer.finish(&self.timeline, ctx).await?;
 
-        self.new_images.push(image_layer);
+        let keyspace = KeySpace {
+            ranges: self.get_keyspace(key_range, lsn, ctx).await?,
+        };
+        // TODO set proper (stateful) start. The create_image_layer_for_rel_blocks function mostly
+        let start = Key::MIN;
+        let ImageLayerCreationOutcome {
+            image,
+            next_start_key: _,
+        } = self
+            .timeline
+            .create_image_layer_for_rel_blocks(
+                &keyspace,
+                image_layer_writer,
+                lsn,
+                ctx,
+                key_range.clone(),
+                start,
+            )
+            .await?;
+
+        if let Some(image_layer) = image {
+            self.new_images.push(image_layer);
+        }
 
         timer.stop_and_record();
 

pageserver/src/tenant/timeline/layer_manager.rs

@@ -166,7 +166,7 @@ impl LayerManager {
     /// Flush a frozen layer and add the written delta layer to the layer map.
     pub(crate) fn finish_flush_l0_layer(
         &mut self,
-        delta_layer: Option<&ResidentLayer>,
+        delta_layers: &[ResidentLayer],
         frozen_layer_for_check: &Arc<InMemoryLayer>,
         metrics: &TimelineMetrics,
     ) {
@@ -181,10 +181,12 @@ impl LayerManager {
         // layer to disk at the same time, that would not work.
         assert_eq!(Arc::as_ptr(&inmem), Arc::as_ptr(frozen_layer_for_check));
 
-        if let Some(l) = delta_layer {
+        if !delta_layers.is_empty() {
             let mut updates = self.layer_map.batch_update();
-            Self::insert_historic_layer(l.as_ref().clone(), &mut updates, &mut self.layer_fmgr);
-            metrics.record_new_file_metrics(l.layer_desc().file_size);
+            for l in delta_layers {
+                Self::insert_historic_layer(l.as_ref().clone(), &mut updates, &mut self.layer_fmgr);
+                metrics.record_new_file_metrics(l.layer_desc().file_size);
+            }
             updates.flush();
         }
     }

pageserver/src/walingest.rs

@@ -373,7 +373,6 @@ impl WalIngest {
                 if info == pg_constants::XLOG_RUNNING_XACTS {
                     let xlrec = crate::walrecord::XlRunningXacts::decode(&mut buf);
                     self.checkpoint.oldestActiveXid = xlrec.oldest_running_xid;
-                    self.checkpoint_modified = true;
                 }
             }
             _x => {

pgxn/neon_rmgr/neon_rmgr.c

@@ -81,145 +81,10 @@ neon_rm_redo(XLogReaderState *record)
 	}
 }
 
-/*
- * FIXME: This is an odd place for RestoreRunningXactsFromClog(). It's
- * handy to call it from neon_rm_startup(). Move this somewhere else
- * in pgxn/.
- */
-
-#include "access/clog.h"
-#include "access/transam.h"
-#include "access/twophase.h"
-#include "common/controldata_utils.h"
-#include "storage/proc.h"
-#include "storage/procarray.h"
-#include "storage/standby.h"
-#include "catalog/pg_control.h"
-
-/*
- * This is taken from procarry.c
- * TODO: should we better move it to some header file?
- */
-#define PROCARRAY_MAXPROCS	(MaxBackends + max_prepared_xacts)
-#define TOTAL_MAX_CACHED_SUBXIDS \
-	((PGPROC_MAX_CACHED_SUBXIDS + 1) * PROCARRAY_MAXPROCS)
-
-/*
- * We do not want to wait running-xacts record from primary which is
- * generated each 15 seconds (if there is some activity).  So we try
- * to restore information about running transactions from CLOG.
- */
-static void
-RestoreRunningXactsFromClog(void)
-{
-	TransactionId from;
-	TransactionId till;
-	int			xcnt;
-	TransactionId *xids;
-	RunningTransactionsData running;
-	TransactionId latestCompletedXid;
-	int			max_xcnt;
-	bool		overflowed;
-
-	{
-		ControlFileData *ControlFile;
-		bool		crc_ok;
-
-		LWLockAcquire(ControlFileLock, LW_SHARED);
-		ControlFile = get_controlfile(DataDir, &crc_ok);
-		from = ControlFile->checkPointCopy.oldestActiveXid;
-		till = XidFromFullTransactionId(ShmemVariableCache->nextXid);
-		LWLockRelease(ControlFileLock);
-		pfree(ControlFile);
-	}
-
-	if (!TransactionIdIsNormal(from))
-	{
-		/*
-		 * No checkpoint or running-xacts record was written,
-		 * so use most conservative approximation for oldestActiveXid: firstNormalTransactionId.
-		 * There are should not be problems with wraparounf because it is not possible that
-		 * XID is overflown without writting any checkpoint or running-xact record.
-		 */
-		from = FirstNormalTransactionId;
-	}
-
-	/*
-	 * To avoid "too many KnownAssignedXids" error later during replay, we
-	 * limit number of reported transactions. This is a tradeoff: if we are
-	 * willing to consume more of the KnownAssignedXids space for the XIDs
-	 * now, that allows us to start up, but we might run out of space later.
-	 *
-	 * XXX: What would be the safe limit that would guarantee that we won't
-	 * run out of space later? And how much above that are we willing to
-	 * "overbook" hoping that we won't need the space later?
-	 */
-	max_xcnt = TOTAL_MAX_CACHED_SUBXIDS / 2;
-
-	xids = (TransactionId *) palloc(max_xcnt * sizeof(TransactionId));
-	xcnt = 0;
-	for (TransactionId xid = from; TransactionIdPrecedes(xid, till);)
-	{
-		XLogRecPtr	xidlsn;
-		XidStatus	xidstatus = TransactionIdGetStatus(xid, &xidlsn);
-
-		if (xidstatus == TRANSACTION_STATUS_IN_PROGRESS)
-		{
-			if (xcnt < max_xcnt)
-				xids[xcnt] = xid;
-			{
-				/*
-				 * Overflowed. We won't be able to install the
-				 * RunningTransactions snapshot. We could bail out now, but
-				 * keep going to count the XIDs, for the sake of the LOG
-				 * message at the end of the function.
-				 */
-			}
-			xcnt++;
-		}
-		TransactionIdAdvance(xid);
-	}
-
-	/*
-	 * Construct a RunningTransactions snapshot with the XIDs scanned from
-	 * CLOG.
-	 *
-	 * TODO: test that this works right with prepared transactions.
-	 */
-	if (xcnt <= max_xcnt)
-	{
-		running.xcnt = xcnt;
-		running.subxcnt = 0;
-		running.subxid_overflow = false;
-		running.nextXid = XidFromFullTransactionId(ShmemVariableCache->nextXid);
-		running.oldestRunningXid = from;
-		latestCompletedXid = till;
-		TransactionIdRetreat(latestCompletedXid);
-		Assert(TransactionIdIsNormal(latestCompletedXid));
-		running.latestCompletedXid = latestCompletedXid;
-		running.xids = xids;
-
-		ProcArrayApplyRecoveryInfo(&running);
-
-		StandbyRecoverPreparedTransactions();
-
-		elog(LOG, "initialized known-assigned XIDs with %d in-progress XIDs between %u and %u (max %d)",
-			 xcnt, from, till, max_xcnt);
-	}
-	else
-		elog(LOG, "could not initialize known-assigned XIDs because there are too many (sub)transactions active (%d in-progress XIDs between %u and %u, max %d)",
-			 xcnt, from, till, max_xcnt);
-
-	pfree(xids);
-}
-
 static void
 neon_rm_startup(void)
 {
-	if (standbyState == STANDBY_INITIALIZED)
-		RestoreRunningXactsFromClog();
-	else
-	  elog(LOG, "neon_rm_startup called with standbyState=%d", standbyState);
+	/* nothing to do here */
 }
 
 static void

proxy/src/context/parquet.rs

@@ -355,7 +355,7 @@ async fn upload_parquet(
         "{year:04}/{month:02}/{day:02}/{hour:02}/requests_{id}.parquet"
     ))?;
     let cancel = CancellationToken::new();
-    backoff::retry(
+    let maybe_err = backoff::retry(
         || async {
             let stream = futures::stream::once(futures::future::ready(Ok(data.clone())));
             storage
@@ -372,7 +372,12 @@ async fn upload_parquet(
     .await
     .ok_or_else(|| anyhow::Error::new(TimeoutOrCancel::Cancel))
     .and_then(|x| x)
-    .context("request_data_upload")?;
+    .context("request_data_upload")
+    .err();
+
+    if let Some(err) = maybe_err {
+        tracing::warn!(%id, %err, "failed to upload request data");
+    }
 
     Ok(buffer.writer())
 }

safekeeper/src/safekeeper.rs

@@ -827,10 +827,10 @@ where
 
     /// Persist control file if there is something to save and enough time
     /// passed after the last save.
-    pub async fn maybe_persist_inmem_control_file(&mut self) -> Result<()> {
+    pub async fn maybe_persist_inmem_control_file(&mut self) -> Result<bool> {
         const CF_SAVE_INTERVAL: Duration = Duration::from_secs(300);
         if self.state.pers.last_persist_at().elapsed() < CF_SAVE_INTERVAL {
-            return Ok(());
+            return Ok(false);
         }
         let need_persist = self.state.inmem.commit_lsn > self.state.commit_lsn
             || self.state.inmem.backup_lsn > self.state.backup_lsn
@@ -840,7 +840,7 @@ where
             self.state.flush().await?;
             trace!("saved control file: {CF_SAVE_INTERVAL:?} passed");
         }
-        Ok(())
+        Ok(need_persist)
     }
 
     /// Handle request to append WAL.

safekeeper/src/timeline.rs

@@ -104,11 +104,16 @@ pub type ReadGuardSharedState<'a> = RwLockReadGuard<'a, SharedState>;
 pub struct WriteGuardSharedState<'a> {
     tli: Arc<Timeline>,
     guard: RwLockWriteGuard<'a, SharedState>,
+    skip_update: bool,
 }
 
 impl<'a> WriteGuardSharedState<'a> {
     fn new(tli: Arc<Timeline>, guard: RwLockWriteGuard<'a, SharedState>) -> Self {
-        WriteGuardSharedState { tli, guard }
+        WriteGuardSharedState {
+            tli,
+            guard,
+            skip_update: false,
+        }
     }
 }
 
@@ -149,10 +154,12 @@ impl<'a> Drop for WriteGuardSharedState<'a> {
             }
         });
 
-        // send notification about shared state update
-        self.tli.shared_state_version_tx.send_modify(|old| {
-            *old += 1;
-        });
+        if !self.skip_update {
+            // send notification about shared state update
+            self.tli.shared_state_version_tx.send_modify(|old| {
+                *old += 1;
+            });
+        }
     }
 }
 
@@ -802,7 +809,11 @@ impl Timeline {
 
         // update last_removed_segno
         let mut shared_state = self.write_shared_state().await;
-        shared_state.last_removed_segno = horizon_segno;
+        if shared_state.last_removed_segno != horizon_segno {
+            shared_state.last_removed_segno = horizon_segno;
+        } else {
+            shared_state.skip_update = true;
+        }
         Ok(())
     }
 
@@ -811,11 +822,10 @@ impl Timeline {
     /// to date so that storage nodes restart doesn't cause many pageserver ->
     /// safekeeper reconnections.
     pub async fn maybe_persist_control_file(self: &Arc<Self>) -> Result<()> {
-        self.write_shared_state()
-            .await
-            .sk
-            .maybe_persist_inmem_control_file()
-            .await
+        let mut guard = self.write_shared_state().await;
+        let changed = guard.sk.maybe_persist_inmem_control_file().await?;
+        guard.skip_update = !changed;
+        Ok(())
     }
 
     /// Gather timeline data for metrics.

test_runner/fixtures/pageserver/allowed_errors.py

@@ -70,7 +70,7 @@ DEFAULT_PAGESERVER_ALLOWED_ERRORS = (
     # this is expected given our collaborative shutdown approach for the UploadQueue
     ".*Compaction failed.*, retrying in .*: Other\\(queue is in state Stopped.*",
     ".*Compaction failed.*, retrying in .*: ShuttingDown",
-    ".*Compaction failed.*, retrying in .*: timeline shutting down.*",
+    ".*Compaction failed.*, retrying in .*: Other\\(timeline shutting down.*",
     # Pageserver timeline deletion should be polled until it gets 404, so ignore it globally
     ".*Error processing HTTP request: NotFound: Timeline .* was not found",
     ".*took more than expected to complete.*",

test_runner/fixtures/utils.py

@@ -541,11 +541,22 @@ def assert_pageserver_backups_equal(left: Path, right: Path, skip_files: Set[str
 
     left_list, right_list = map(build_hash_list, [left, right])
 
-    try:
-        assert len(left_list) == len(right_list)
+    assert len(left_list) == len(
+        right_list
+    ), f"unexpected number of files on tar files, {len(left_list)} != {len(right_list)}"
 
-        for left_tuple, right_tuple in zip(left_list, right_list):
-            assert left_tuple == right_tuple
-    finally:
-        elapsed = time.time() - started_at
-        log.info(f"assert_pageserver_backups_equal completed in {elapsed}s")
+    mismatching = set()
+
+    for left_tuple, right_tuple in zip(left_list, right_list):
+        left_path, left_hash = left_tuple
+        right_path, right_hash = right_tuple
+        assert (
+            left_path == right_path
+        ), f"file count matched, expected these to be same paths: {left_path}, {right_path}"
+        if left_hash != right_hash:
+            mismatching.add(left_path)
+
+    assert len(mismatching) == 0, f"files with hash mismatch: {mismatching}"
+
+    elapsed = time.time() - started_at
+    log.info(f"assert_pageserver_backups_equal completed in {elapsed}s")

test_runner/regress/test_attach_tenant_config.py

@@ -17,9 +17,13 @@ def positive_env(neon_env_builder: NeonEnvBuilder) -> NeonEnv:
     neon_env_builder.enable_pageserver_remote_storage(RemoteStorageKind.LOCAL_FS)
     env = neon_env_builder.init_start()
 
-    # eviction might be the first one after an attach to access the layers
-    env.pageserver.allowed_errors.append(
-        ".*unexpectedly on-demand downloading remote layer .* for task kind Eviction"
+    env.pageserver.allowed_errors.extend(
+        [
+            # eviction might be the first one after an attach to access the layers
+            ".*unexpectedly on-demand downloading remote layer .* for task kind Eviction",
+            # detach can happen before we get to validate the generation number
+            ".*deletion backend: Dropped remote consistent LSN updates for tenant.*",
+        ]
     )
     assert isinstance(env.pageserver_remote_storage, LocalFsStorage)
     return env

test_runner/regress/test_import.py

@@ -163,7 +163,7 @@ def test_import_from_pageserver_small(
 
     num_rows = 3000
     lsn = _generate_data(num_rows, endpoint)
-    _import(num_rows, lsn, env, pg_bin, timeline, env.pg_distrib_dir, test_output_dir)
+    _import(num_rows, lsn, env, pg_bin, timeline, test_output_dir)
 
 
 @pytest.mark.timeout(1800)
@@ -193,9 +193,7 @@ def test_import_from_pageserver_multisegment(
     log.info(f"timeline logical size = {logical_size / (1024 ** 2)}MB")
     assert logical_size > 1024**3  # = 1GB
 
-    tar_output_file = _import(
-        num_rows, lsn, env, pg_bin, timeline, env.pg_distrib_dir, test_output_dir
-    )
+    tar_output_file = _import(num_rows, lsn, env, pg_bin, timeline, test_output_dir)
 
     # Check if the backup data contains multiple segment files
     cnt_seg_files = 0
@@ -235,7 +233,6 @@ def _import(
     env: NeonEnv,
     pg_bin: PgBin,
     timeline: TimelineId,
-    pg_distrib_dir: Path,
     test_output_dir: Path,
 ) -> Path:
     """Test importing backup data to the pageserver.
@@ -295,7 +292,7 @@ def _import(
     wait_for_upload(client, tenant, timeline, lsn)
 
     # Check it worked
-    endpoint = env.endpoints.create_start(endpoint_id, tenant_id=tenant)
+    endpoint = env.endpoints.create_start(endpoint_id, tenant_id=tenant, lsn=lsn)
     assert endpoint.safe_psql("select count(*) from tbl") == [(expected_num_rows,)]
 
     # Take another fullbackup

test_runner/regress/test_replication_start.py

@@ -1,7 +1,9 @@
+import pytest
 from fixtures.log_helper import log
 from fixtures.neon_fixtures import NeonEnv, wait_replica_caughtup
 
 
+@pytest.mark.xfail
 def test_replication_start(neon_simple_env: NeonEnv):
     env = neon_simple_env
 

test_runner/regress/test_replication_start_subxid_overflow.py

@@ -1,239 +0,0 @@
-from fixtures.log_helper import log
-from fixtures.neon_fixtures import NeonEnv, wait_for_last_flush_lsn, wait_replica_caughtup
-
-
-# This test replica startup in case of large number of active subtransactions at primary node.
-# Number of known transaction xids is limited and if we do not limit number of transactions
-# restored from CLOG, then it will cause crash of replica.
-def test_replication_start_subxid_overflow(neon_simple_env: NeonEnv):
-    env = neon_simple_env
-
-    with env.endpoints.create_start(branch_name="main", endpoint_id="primary") as primary:
-        with primary.connect() as p_con:
-            with p_con.cursor() as p_cur:
-                p_cur.execute("begin")
-                p_cur.execute("create table t(pk serial primary key, payload integer)")
-                p_cur.execute(
-                    """create or replace function create_subxacts(n integer) returns void as $$
-                    declare
-                        i integer;
-                    begin
-                       for i in 1..n loop
-                          begin
-                              insert into t (payload) values (0);
-                          exception
-                              when others then
-                                  raise exception 'caught something';
-                          end;
-                       end loop;
-                   end; $$ language plpgsql"""
-                )
-                p_cur.execute("select create_subxacts(100000)")
-                p_cur.execute("select txid_current()")
-                xid = p_cur.fetchall()[0][0]
-                log.info(f"Master transaction {xid}")
-                with env.endpoints.new_replica_start(
-                    origin=primary, endpoint_id="secondary"
-                ) as secondary:
-                    wait_replica_caughtup(primary, secondary)
-                    with secondary.connect() as s_con:
-                        with s_con.cursor() as s_cur:
-                            # Enforce setting hint bits for pg_class tuples.
-                            # If master's transaction is not marked as in-progress in MVCC snapshot,
-                            # then XMIN_INVALID hint bit will be set for table's 't' tuple makeing it invisible.
-                            s_cur.execute("select * from pg_class")
-                            p_cur.execute("commit")
-                            wait_replica_caughtup(primary, secondary)
-                            s_cur.execute("select * from t where pk = 1")
-                            assert s_cur.fetchone() == (1, 0)
-
-
-# Test starting a replica at a point in time where there is a transaction
-# running in primary with lots of subtransactions. Test MVCC in standby before
-# and after the commit of the transaction.
-#
-# XXX: This currently fails, the query in standby incorrectly sees some sub-xids
-# as committed:
-#
-# test_runner/regress/test_replication_start_subxid_overflow.py:105: AssertionError
-#
-# FAILED test_runner/regress/test_replication_start_subxid_overflow.py::test_replication_start_subxid_overflow2[debug-pg16] - assert (93229,) == (0,)
-
-def test_replication_start_subxid_overflow2(neon_simple_env: NeonEnv):
-    env = neon_simple_env
-
-    primary = env.endpoints.create_start(branch_name="main", endpoint_id="primary")
-
-    p_con = primary.connect()
-    p_cur = p_con.cursor()
-
-    p_cur.execute("create table t(pk serial primary key, payload integer)")
-    p_cur.execute(
-        """create or replace function create_subxacts(n integer) returns void as $$
-        declare
-            i integer;
-        begin
-           for i in 1..n loop
-              begin
-                  insert into t (payload) values (0);
-              exception
-                  when others then
-                      raise exception 'caught something';
-              end;
-           end loop;
-       end; $$ language plpgsql"""
-    )
-
-    # Start a new transaction in primary, with a lot of subtransactions
-    p_cur.execute("begin")
-    p_cur.execute("select create_subxacts(100000)")
-
-    # Create a replica at this LSN
-    wait_for_last_flush_lsn(env, primary, env.initial_tenant, env.initial_timeline)
-    with env.endpoints.new_replica_start(origin=primary, endpoint_id="secondary") as secondary:
-        s_con = secondary.connect()
-        s_cur = s_con.cursor()
-
-        # The transaction in primary has not committed yet.
-        wait_replica_caughtup(primary, secondary)
-        s_cur.execute("select count(*) from t")
-        assert s_cur.fetchone() == (0,)
-
-        # Add more sub-xids to the same transaction in primary
-        p_cur.execute("select create_subxacts(10000)")
-
-        # The transaction still hasn't committed
-        wait_replica_caughtup(primary, secondary)
-        s_cur.execute("BEGIN ISOLATION LEVEL REPEATABLE READ")
-        s_cur.execute("select count(*) from t")
-        assert s_cur.fetchone() == (0,)
-
-        # Commit the transaction in primary
-        p_cur.execute("commit")
-
-        # Should still be invisible to the old snapshot
-        wait_replica_caughtup(primary, secondary)
-        s_cur.execute("select count(*) from t")
-        assert s_cur.fetchone() == (0,)
-
-        # Commit the REPEATABLE READ transaction in standby
-        s_cur.execute("commit")
-
-        # The transaction should now visible to new snapshot in standby
-        s_cur.execute("select count(*) from t")
-        assert s_cur.fetchone() == (200000,)
-
-# Test starting a standby from situation that there is one transaction running
-# in primary, with lots of subtransactions. Then, after the standby is running,
-# start a large number of new transcations, with lots of subtransactions each, in
-# the primary.
-#
-# XXX: This currently crashes the startup process in secondary with:
-#
-# FATAL:  too many KnownAssignedXids
-# CONTEXT:  WAL redo at 0/1895CB0 for neon/INSERT: off: 25, flags: 0x08; blkref #0: rel 1663/5/16385, blk 64
-def test_replication_start_subxid_overflow3(neon_simple_env: NeonEnv):
-    env = neon_simple_env
-
-    primary = env.endpoints.create_start(branch_name="main", endpoint_id="primary")
-
-    with primary.connect() as p_con:
-        p_cur = p_con.cursor()
-
-        p_cur.execute("create table t(pk serial primary key, payload integer)")
-        p_cur.execute(
-            """create or replace function create_subxacts(n integer) returns void as $$
-            declare
-                i integer;
-            begin
-                for i in 1..n loop
-                    begin
-                        insert into t (payload) values (0);
-                    exception
-                        when others then
-                            raise exception 'caught something';
-                    end;
-                end loop;
-           end; $$ language plpgsql"""
-        )
-
-        p_cur.execute(f"show max_connections")
-        max_connections = int(p_cur.fetchone()[0])
-
-    n_connections = max_connections - 2
-    n_subxids = 100
-
-    # Start one top tranaction in primary, with lots of subtransactions. This uses up much the
-    # known-assigned XIDs space in the standby, but doesn't cause it to overflow.
-    large_p_conn = primary.connect()
-    large_p_cur = large_p_conn.cursor()
-    large_p_cur.execute("begin")
-    large_p_cur.execute("select create_subxacts(2000)")
-
-    # Create a replica at this LSN
-    wait_for_last_flush_lsn(env, primary, env.initial_tenant, env.initial_timeline)
-    with env.endpoints.new_replica_start(origin=primary, endpoint_id="secondary") as secondary:
-        s_con = secondary.connect()
-        s_cur = s_con.cursor()
-
-        # The transaction in primary has not committed yet.
-        wait_replica_caughtup(primary, secondary)
-        s_cur.execute("select count(*) from t")
-        assert s_cur.fetchone() == (0,)
-
-        # Start max number of top transactions in primary, with a lot of
-        # subtransactions each We add the subtransactions to each top
-        # transaction in a round-robin fashion, instead of adding a lot of
-        # subtransactions ot one top transaction at a time. This way, we will
-        # have the max number of subtransactions in the in-memory subxid cache
-        # of each top transaction, until they all overflow.
-        #
-        # Currently, PGPROC_MAX_CACHED_SUBXIDS == 64, so this will overflow the
-        # all the subxid caches after creating 64 subxids in each top
-        # transaction. The point just before the caches have overflowed is the
-        # most interesting point in time, but we'll keep going beyond that, to
-        # ensure that this test is robust even if PGPROC_MAX_CACHED_SUBXIDS
-        # changes.
-        p_cons = []
-        p_curs = []
-        for i in range(0, n_connections):
-            p_con = primary.connect()
-            p_cur = p_con.cursor()
-
-            p_cur.execute("begin")
-            p_cons.append(p_con)
-            p_curs.append(p_cur)
-
-        for subxid in range(0, n_subxids):
-            for i in range(0, n_connections):
-                p_curs[i].execute("select create_subxacts(1)")
-
-            # None of the transactions have committed yet, so they should be
-            # invisible in standby.
-            s_cur.execute("select count(*) from t")
-            assert s_cur.fetchone() == (0,)
-
-        # The transactions still haven't committed. Start a new snapshot, we
-        # will use it later, after committing in the primary.
-        wait_replica_caughtup(primary, secondary)
-        s_cur.execute("BEGIN ISOLATION LEVEL REPEATABLE READ")
-
-        s_cur.execute("select count(*) from t")
-        assert s_cur.fetchone() == (0,)
-
-        # Commit all the transactions in the primary
-        large_p_cur.execute("commit")
-        for i in range(0, n_connections):
-            p_curs[i].execute("commit")
-
-        # All the XIDs still be invisible to the old snapshot
-        wait_replica_caughtup(primary, secondary)
-        s_cur.execute("select count(*) from t")
-        assert s_cur.fetchone() == (0,)
-
-        # Commit the REPEATABLE READ transaction in standby
-        s_cur.execute("commit")
-
-        # The transaction should now visible to new snapshot in standby
-        s_cur.execute("select count(*) from t")
-        assert s_cur.fetchone() == (n_connections * n_subxids,)

vendor/revisions.json

@@ -1,5 +1,5 @@
 {
-  "v16": ["16.3", "c83756da5260c784f68ca21c037c7e3fd4ca87c4"],
-  "v15": ["15.7", "2712558896d95fcf27963da6c64a5f844d0621fa"],
-  "v14": ["14.12", "05d21e2b4167245102eddee01a158264fb2eddfe"]
+  "v16": ["16.3", "3c2b9d576c580e0b5b7108001f959b8c5b42e0a2"],
+  "v15": ["15.7", "74fb144890c4f955db1ef50ee1eeb9d8a6c2f69d"],
+  "v14": ["14.12", "0d30e28f74f49fe6a27a6bd45dcfeb1060656b8f"]
 }