build(deps): bump certifi from 2023.7.22 to 2024.7.4 (#8301)

The find-large-objects scrubber subcommand is quite fast if you run it
in an environment with low latency to the S3 bucket (say an EC2 instance
in the same region). However, the higher the latency gets, the slower
the command becomes. Therefore, add a concurrency param and make it
parallelized. This doesn't change that general relationship, but at
least lets us do multiple requests in parallel and therefore hopefully
faster.

Running with concurrency of 64 (default):

```
2024-07-05T17:30:22.882959Z  INFO lazy_load_identity [...]
[...]
2024-07-05T17:30:28.289853Z  INFO Scanned 500 shards. [...]
```

With concurrency of 1, simulating state before this PR:

```
2024-07-05T17:31:43.375153Z  INFO lazy_load_identity [...]
[...]
2024-07-05T17:33:51.987092Z  INFO Scanned 500 shards. [...]
```

In other words, to list 500 shards, speed is increased from 2:08 minutes
to 6 seconds.

Follow-up of  #8257, part of #5431

pageserver/src/metrics.rs

@@ -500,15 +500,6 @@ static RESIDENT_PHYSICAL_SIZE: Lazy<UIntGaugeVec> = Lazy::new(|| {
     .expect("failed to define a metric")
 });
 
-static VISIBLE_PHYSICAL_SIZE: Lazy<UIntGaugeVec> = Lazy::new(|| {
-    register_uint_gauge_vec!(
-        "pageserver_visible_physical_size",
-        "The size of the layer files present in the pageserver's filesystem.",
-        &["tenant_id", "shard_id", "timeline_id"]
-    )
-    .expect("failed to define a metric")
-});
-
 pub(crate) static RESIDENT_PHYSICAL_SIZE_GLOBAL: Lazy<UIntGauge> = Lazy::new(|| {
     register_uint_gauge!(
         "pageserver_resident_physical_size_global",
@@ -2139,7 +2130,6 @@ pub(crate) struct TimelineMetrics {
     pub archival_size: UIntGauge,
     pub standby_horizon_gauge: IntGauge,
     pub resident_physical_size_gauge: UIntGauge,
-    pub visible_physical_size_gauge: UIntGauge,
     /// copy of LayeredTimeline.current_logical_size
     pub current_logical_size_gauge: UIntGauge,
     pub aux_file_size_gauge: IntGauge,
@@ -2226,9 +2216,6 @@ impl TimelineMetrics {
         let resident_physical_size_gauge = RESIDENT_PHYSICAL_SIZE
             .get_metric_with_label_values(&[&tenant_id, &shard_id, &timeline_id])
             .unwrap();
-        let visible_physical_size_gauge = VISIBLE_PHYSICAL_SIZE
-            .get_metric_with_label_values(&[&tenant_id, &shard_id, &timeline_id])
-            .unwrap();
         // TODO: we shouldn't expose this metric
         let current_logical_size_gauge = CURRENT_LOGICAL_SIZE
             .get_metric_with_label_values(&[&tenant_id, &shard_id, &timeline_id])
@@ -2279,7 +2266,6 @@ impl TimelineMetrics {
             archival_size,
             standby_horizon_gauge,
             resident_physical_size_gauge,
-            visible_physical_size_gauge,
             current_logical_size_gauge,
             aux_file_size_gauge,
             directory_entries_count_gauge,
@@ -2331,7 +2317,6 @@ impl TimelineMetrics {
             RESIDENT_PHYSICAL_SIZE_GLOBAL.sub(self.resident_physical_size_get());
             let _ = RESIDENT_PHYSICAL_SIZE.remove_label_values(&[tenant_id, shard_id, timeline_id]);
         }
-        let _ = VISIBLE_PHYSICAL_SIZE.remove_label_values(&[tenant_id, shard_id, timeline_id]);
         let _ = CURRENT_LOGICAL_SIZE.remove_label_values(&[tenant_id, shard_id, timeline_id]);
         if let Some(metric) = Lazy::get(&DIRECTORY_ENTRIES_COUNT) {
             let _ = metric.remove_label_values(&[tenant_id, shard_id, timeline_id]);

pageserver/src/pgdatadir_mapping.rs

@@ -931,7 +931,7 @@ impl Timeline {
             result.to_keyspace(),
             /* AUX sparse key space */
             SparseKeySpace(KeySpace {
-                ranges: vec![Key::metadata_aux_key_range(), repl_origin_key_range()],
+                ranges: vec![repl_origin_key_range(), Key::metadata_aux_key_range()],
             }),
         ))
     }

pageserver/src/tenant.rs

@@ -19,7 +19,6 @@ use enumset::EnumSet;
 use futures::stream::FuturesUnordered;
 use futures::FutureExt;
 use futures::StreamExt;
-use pageserver_api::keyspace::KeySpace;
 use pageserver_api::models;
 use pageserver_api::models::AuxFilePolicy;
 use pageserver_api::models::TimelineState;
@@ -31,7 +30,6 @@ use pageserver_api::shard::TenantShardId;
 use remote_storage::DownloadError;
 use remote_storage::GenericRemoteStorage;
 use remote_storage::TimeoutOrCancel;
-use std::collections::BTreeMap;
 use std::fmt;
 use std::time::SystemTime;
 use storage_broker::BrokerClientChannel;
@@ -94,12 +92,14 @@ use crate::tenant::storage_layer::ImageLayer;
 use crate::walredo;
 use crate::InitializationOrder;
 use std::collections::hash_map::Entry;
+use std::collections::BTreeSet;
 use std::collections::HashMap;
 use std::collections::HashSet;
 use std::fmt::Debug;
 use std::fmt::Display;
 use std::fs;
 use std::fs::File;
+use std::ops::Bound::Included;
 use std::sync::atomic::AtomicU64;
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
@@ -632,11 +632,6 @@ impl Tenant {
                     timeline.maybe_spawn_flush_loop();
                 }
             }
-
-            if let Some(ancestor) = timeline.get_ancestor_timeline() {
-                let mut ancestor_gc_info = ancestor.gc_info.write().unwrap();
-                ancestor_gc_info.insert_child(timeline.timeline_id, timeline.get_ancestor_lsn());
-            }
         };
 
         // Sanity check: a timeline should have some content.
@@ -1738,9 +1733,6 @@ impl Tenant {
                 .values()
                 .filter(|timeline| !(timeline.is_broken() || timeline.is_stopping()));
 
-            // Before activation, populate each Timeline's GcInfo with information about its children
-            self.initialize_gc_info(&timelines_accessor);
-
             // Spawn gc and compaction loops. The loops will shut themselves
             // down when they notice that the tenant is inactive.
             tasks::start_background_loops(self, background_jobs_can_start);
@@ -2773,56 +2765,6 @@ impl Tenant {
             .await
     }
 
-    /// Populate all Timelines' `GcInfo` with information about their children.  We do not set the
-    /// PITR cutoffs here, because that requires I/O: this is done later, before GC, by [`Self::refresh_gc_info_internal`]
-    ///
-    /// Subsequently, parent-child relationships are updated incrementally during timeline creation/deletion.
-    fn initialize_gc_info(
-        &self,
-        timelines: &std::sync::MutexGuard<HashMap<TimelineId, Arc<Timeline>>>,
-    ) {
-        // This function must be called before activation: after activation timeline create/delete operations
-        // might happen, and this function is not safe to run concurrently with those.
-        assert!(!self.is_active());
-
-        // Scan all timelines. For each timeline, remember the timeline ID and
-        // the branch point where it was created.
-        let mut all_branchpoints: BTreeMap<TimelineId, Vec<(Lsn, TimelineId, Option<KeySpace>)>> =
-            BTreeMap::new();
-        timelines.iter().for_each(|(timeline_id, timeline_entry)| {
-            if let Some(ancestor_timeline_id) = &timeline_entry.get_ancestor_timeline_id() {
-                let ancestor_children = all_branchpoints.entry(*ancestor_timeline_id).or_default();
-                ancestor_children.push((timeline_entry.get_ancestor_lsn(), *timeline_id, None));
-            }
-        });
-
-        // The number of bytes we always keep, irrespective of PITR: this is a constant across timelines
-        let horizon = self.get_gc_horizon();
-
-        // Populate each timeline's GcInfo with information about its child branches
-        for timeline in timelines.values() {
-            let mut branchpoints: Vec<(Lsn, TimelineId, Option<KeySpace>)> = all_branchpoints
-                .remove(&timeline.timeline_id)
-                .unwrap_or_default();
-
-            branchpoints.sort_by_key(|b| b.0);
-
-            let mut target = timeline.gc_info.write().unwrap();
-
-            target.retain_lsns = branchpoints;
-
-            let horizon_cutoff = timeline
-                .get_last_record_lsn()
-                .checked_sub(horizon)
-                .unwrap_or(Lsn(0));
-
-            target.cutoffs = GcCutoffs {
-                horizon: horizon_cutoff,
-                pitr: Lsn::INVALID,
-            };
-        }
-    }
-
     async fn refresh_gc_info_internal(
         &self,
         target_timeline_id: Option<TimelineId>,
@@ -2845,11 +2787,6 @@ impl Tenant {
             .cloned()
             .collect::<Vec<_>>();
 
-        if target_timeline_id.is_some() && timelines.is_empty() {
-            // We were to act on a particular timeline and it wasn't found
-            return Err(GcError::TimelineNotFound);
-        }
-
         let mut gc_cutoffs: HashMap<TimelineId, GcCutoffs> =
             HashMap::with_capacity(timelines.len());
 
@@ -2872,15 +2809,68 @@ impl Tenant {
         // because that will stall branch creation.
         let gc_cs = self.gc_cs.lock().await;
 
+        // Scan all timelines. For each timeline, remember the timeline ID and
+        // the branch point where it was created.
+        let (all_branchpoints, timelines): (BTreeSet<(TimelineId, Lsn)>, _) = {
+            let timelines = self.timelines.lock().unwrap();
+            let mut all_branchpoints = BTreeSet::new();
+            let timelines = {
+                if let Some(target_timeline_id) = target_timeline_id.as_ref() {
+                    if timelines.get(target_timeline_id).is_none() {
+                        return Err(GcError::TimelineNotFound);
+                    }
+                };
+
+                timelines
+                    .iter()
+                    .map(|(_timeline_id, timeline_entry)| {
+                        if let Some(ancestor_timeline_id) =
+                            &timeline_entry.get_ancestor_timeline_id()
+                        {
+                            // If target_timeline is specified, we only need to know branchpoints of its children
+                            if let Some(timeline_id) = target_timeline_id {
+                                if ancestor_timeline_id == &timeline_id {
+                                    all_branchpoints.insert((
+                                        *ancestor_timeline_id,
+                                        timeline_entry.get_ancestor_lsn(),
+                                    ));
+                                }
+                            }
+                            // Collect branchpoints for all timelines
+                            else {
+                                all_branchpoints.insert((
+                                    *ancestor_timeline_id,
+                                    timeline_entry.get_ancestor_lsn(),
+                                ));
+                            }
+                        }
+
+                        timeline_entry.clone()
+                    })
+                    .collect::<Vec<_>>()
+            };
+            (all_branchpoints, timelines)
+        };
+
         // Ok, we now know all the branch points.
         // Update the GC information for each timeline.
         let mut gc_timelines = Vec::with_capacity(timelines.len());
         for timeline in timelines {
-            // We filtered the timeline list above
+            // If target_timeline is specified, ignore all other timelines
             if let Some(target_timeline_id) = target_timeline_id {
-                assert_eq!(target_timeline_id, timeline.timeline_id);
+                if timeline.timeline_id != target_timeline_id {
+                    continue;
+                }
             }
 
+            let branchpoints: Vec<Lsn> = all_branchpoints
+                .range((
+                    Included((timeline.timeline_id, Lsn(0))),
+                    Included((timeline.timeline_id, Lsn(u64::MAX))),
+                ))
+                .map(|&x| x.1)
+                .collect();
+
             {
                 let mut target = timeline.gc_info.write().unwrap();
 
@@ -2918,12 +2908,20 @@ impl Tenant {
                         .0,
                 );
 
-                // Apply the cutoffs we found to the Timeline's GcInfo.  Why might we _not_ have cutoffs for a timeline?
-                // - this timeline was created while we were finding cutoffs
-                // - lsn for timestamp search fails for this timeline repeatedly
-                if let Some(cutoffs) = gc_cutoffs.remove(&timeline.timeline_id) {
-                    target.cutoffs = cutoffs;
-                }
+                match gc_cutoffs.remove(&timeline.timeline_id) {
+                    Some(cutoffs) => {
+                        target.retain_lsns = branchpoints;
+                        target.cutoffs = cutoffs;
+                    }
+                    None => {
+                        // reasons for this being unavailable:
+                        // - this timeline was created while we were finding cutoffs
+                        // - lsn for timestamp search fails for this timeline repeatedly
+                        //
+                        // in both cases, refreshing the branchpoints is correct.
+                        target.retain_lsns = branchpoints;
+                    }
+                };
             }
 
             gc_timelines.push(timeline);
@@ -4307,7 +4305,7 @@ mod tests {
         {
             let branchpoints = &tline.gc_info.read().unwrap().retain_lsns;
             assert_eq!(branchpoints.len(), 1);
-            assert_eq!(branchpoints[0], (Lsn(0x40), NEW_TIMELINE_ID, None));
+            assert_eq!(branchpoints[0], Lsn(0x40));
         }
 
         // You can read the key from the child branch even though the parent is

pageserver/src/tenant/layer_map.rs

@@ -51,7 +51,7 @@ use crate::keyspace::KeyPartitioning;
 use crate::repository::Key;
 use crate::tenant::storage_layer::InMemoryLayer;
 use anyhow::Result;
-use pageserver_api::keyspace::{KeySpace, KeySpaceAccum, KeySpaceRandomAccum};
+use pageserver_api::keyspace::KeySpaceAccum;
 use std::collections::{HashMap, VecDeque};
 use std::iter::Peekable;
 use std::ops::Range;
@@ -61,7 +61,7 @@ use utils::lsn::Lsn;
 use historic_layer_coverage::BufferedHistoricLayerCoverage;
 pub use historic_layer_coverage::LayerKey;
 
-use super::storage_layer::{LayerVisibility, PersistentLayerDesc};
+use super::storage_layer::PersistentLayerDesc;
 
 ///
 /// LayerMap tracks what layers exist on a timeline.
@@ -870,164 +870,6 @@ impl LayerMap {
         println!("End dump LayerMap");
         Ok(())
     }
-
-    /// `read_points` represent the tip of a timeline and any branch points, i.e. the places
-    /// where we expect to serve reads.
-    ///
-    /// This function is O(N) and should be called infrequently.  The caller is responsible for
-    /// looking up and updating the Layer objects for these layer descriptors.
-    pub(crate) fn get_visibility(
-        &self,
-        mut read_points: Vec<(Lsn, KeySpace)>,
-    ) -> (Vec<(Arc<PersistentLayerDesc>, LayerVisibility)>, KeySpace) {
-        // This is like a KeySpace, but written for efficient subtraction of layers and unions with KeySpaces
-        struct KeyShadow {
-            // FIXME: consider efficiency.  KeySpace is a flat vector, so in principle fairly inefficient for
-            // repeatedly calling contains(), BUT as we iterate through the layermap we expect the shadow to shrink
-            // to something quite small, and for small collections an algorithmically expensive vector is often better
-            // for performance than a more algorithmically cheap data structure.
-            inner: KeySpace,
-        }
-
-        impl KeyShadow {
-            fn new(keyspace: KeySpace) -> Self {
-                Self { inner: keyspace }
-            }
-
-            fn contains(&self, range: Range<Key>) -> bool {
-                self.inner.overlaps(&range)
-            }
-
-            /// Return true if anything was removed.
-            fn subtract(&mut self, range: Range<Key>) -> bool {
-                let removed = self.inner.remove_overlapping_with(&KeySpace {
-                    ranges: vec![range],
-                });
-                !removed.ranges.is_empty()
-            }
-
-            fn union_with(&mut self, keyspace: KeySpace) {
-                let mut accum = KeySpaceRandomAccum::new();
-                let prev = std::mem::take(&mut self.inner);
-                accum.add_keyspace(prev);
-                accum.add_keyspace(keyspace);
-                self.inner = accum.to_keyspace();
-            }
-        }
-
-        // The 'shadow' will be updated as we sweep through the layers: an image layer subtracts from the shadow,
-        // and a ReadPoint
-        read_points.sort_by_key(|rp| rp.0);
-        let mut shadow = KeyShadow::new(
-            read_points
-                .pop()
-                .expect("Every timeline has at least one read point")
-                .1,
-        );
-
-        // We will interleave all our read points and layers into a sorted collection
-        enum Item {
-            ReadPoint { lsn: Lsn, keyspace: KeySpace },
-            Layer(Arc<PersistentLayerDesc>),
-        }
-
-        let mut items = Vec::with_capacity(self.historic.len() + read_points.len());
-        items.extend(self.iter_historic_layers().map(Item::Layer));
-        items.extend(read_points.into_iter().map(|rp| Item::ReadPoint {
-            lsn: rp.0,
-            keyspace: rp.1,
-        }));
-
-        // Ordering: we want to iterate like this:
-        // 1. Highest LSNs first
-        // 2. Consider ReadPoints before image layers if they're at the same LSN
-        items.sort_by_key(|item| {
-            std::cmp::Reverse(match item {
-                Item::ReadPoint {
-                    lsn,
-                    keyspace: _keyspace,
-                } => (*lsn, 0),
-                Item::Layer(layer) => {
-                    if layer.is_delta() {
-                        (layer.get_lsn_range().end, 1)
-                    } else {
-                        (layer.image_layer_lsn(), 2)
-                    }
-                }
-            })
-        });
-
-        let mut results = Vec::with_capacity(self.historic.len());
-
-        // TODO: handle delta layers properly with multiple read points: if a read point intersects a delta layer, we might already
-        // have encountered it and marked it as not-visible.  We need to keep track of which delta layers we are currently within, and
-        // when we encounter a ReadPoint, update the delta layer's visibility as needed.
-        // let mut pending_delta : Vec= ...
-        let mut maybe_covered_deltas: Vec<Arc<PersistentLayerDesc>> = Vec::new();
-
-        for item in items {
-            let (reached_lsn, is_readpoint) = match &item {
-                Item::ReadPoint {
-                    lsn,
-                    keyspace: _keyspace,
-                } => (lsn, true),
-                Item::Layer(layer) => (&layer.lsn_range.start, false),
-            };
-            maybe_covered_deltas.retain(|d| {
-                if *reached_lsn >= d.lsn_range.start && is_readpoint {
-                    // We encountered a readpoint within the delta layer: it is visible
-                    results.push((d.clone(), LayerVisibility::Visible));
-                    false
-                } else if *reached_lsn < d.lsn_range.start {
-                    // We passed the layer's range without encountering a read point: it is not visible
-                    results.push((d.clone(), LayerVisibility::Covered));
-                    false
-                } else {
-                    // We're still in the delta layer: continue iterating
-                    true
-                }
-            });
-
-            match item {
-                Item::ReadPoint {
-                    lsn: _lsn,
-                    keyspace,
-                } => {
-                    shadow.union_with(keyspace);
-                }
-                Item::Layer(layer) => {
-                    let visibility = if layer.is_delta() {
-                        if shadow.contains(layer.get_key_range()) {
-                            LayerVisibility::Visible
-                        } else {
-                            // If a layer isn't visible based on current state, we must defer deciding whether
-                            // it is truly not visible until we have advanced past the delta's range: we might
-                            // encounter another branch point within this delta layer's LSN range.
-                            maybe_covered_deltas.push(layer);
-                            continue;
-                        }
-                    } else if shadow.subtract(layer.get_key_range()) {
-                        // An image layer, which overlapped with the shadow
-                        LayerVisibility::Visible
-                    } else {
-                        // An image layer, which did not overlap with the shadow
-                        LayerVisibility::Covered
-                    };
-
-                    results.push((layer, visibility));
-                }
-            }
-        }
-
-        // Drain any remaining maybe_covered deltas
-        results.extend(
-            maybe_covered_deltas
-                .into_iter()
-                .map(|d| (d, LayerVisibility::Covered)),
-        );
-
-        (results, shadow.inner)
-    }
 }
 
 #[cfg(test)]

pageserver/src/tenant/layer_map/historic_layer_coverage.rs

@@ -521,10 +521,6 @@ impl<Value: Clone> BufferedHistoricLayerCoverage<Value> {
 
         Ok(&self.historic_coverage)
     }
-
-    pub(crate) fn len(&self) -> usize {
-        self.layers.len()
-    }
 }
 
 #[test]

pageserver/src/tenant/size.rs

@@ -271,14 +271,10 @@ pub(super) async fn gather_inputs(
         let mut lsns: Vec<(Lsn, LsnKind)> = gc_info
             .retain_lsns
             .iter()
-            .filter_map(|(lsn, _child_id, _)| {
-                if lsn > &ancestor_lsn {
-                    // this assumes there are no other retain_lsns than the branchpoints
-                    Some((*lsn, LsnKind::BranchPoint))
-                } else {
-                    None
-                }
-            })
+            .filter(|&&lsn| lsn > ancestor_lsn)
+            .copied()
+            // this assumes there are no other retain_lsns than the branchpoints
+            .map(|lsn| (lsn, LsnKind::BranchPoint))
             .collect::<Vec<_>>();
 
         lsns.extend(lease_points.iter().map(|&lsn| (lsn, LsnKind::LeasePoint)));

pageserver/src/tenant/storage_layer.rs

@@ -457,26 +457,6 @@ pub enum ValueReconstructResult {
     Missing,
 }
 
-#[derive(Debug, Clone)]
-pub(crate) enum LayerVisibility {
-    /// A Visible layer might be read while serving a read, because there is not an image layer between it
-    /// and a readable LSN (the tip of the branch or a child's branch point)
-    Visible,
-    /// A Covered layer probably won't be read right now, but _can_ be read in future if someone creates
-    /// a branch or ephemeral endpoint at an LSN below the layer that covers this.
-    Covered,
-    /// Calculating layer visibilty requires I/O, so until this has happened layers are loaded
-    /// in this state.  Note that newly written layers may be called Visible immediately, this uninitialized
-    /// state is for when existing layers are constructed while loading a timeline.
-    Uninitialized,
-}
-
-impl Default for LayerVisibility {
-    fn default() -> Self {
-        Self::Uninitialized
-    }
-}
-
 #[derive(Debug)]
 pub struct LayerAccessStats(Mutex<LayerAccessStatsLocked>);
 
@@ -488,7 +468,6 @@ pub struct LayerAccessStats(Mutex<LayerAccessStatsLocked>);
 struct LayerAccessStatsLocked {
     for_scraping_api: LayerAccessStatsInner,
     for_eviction_policy: LayerAccessStatsInner,
-    visibility: LayerVisibility,
 }
 
 impl LayerAccessStatsLocked {
@@ -612,13 +591,7 @@ impl LayerAccessStats {
             inner.count_by_access_kind[access_kind] += 1;
             inner.task_kind_flag |= ctx.task_kind();
             inner.last_accesses.write(this_access);
-        });
-
-        // We may access a layer marked as Covered, if a new branch was created that depends on
-        // this layer, and background updates to layer visibility didn't notice it yet
-        if !matches!(locked.visibility, LayerVisibility::Visible) {
-            locked.visibility = LayerVisibility::Visible;
-        }
+        })
     }
 
     fn as_api_model(
@@ -700,28 +673,6 @@ impl LayerAccessStats {
             },
         }
     }
-
-    pub(crate) fn set_visibility(&self, visibility: LayerVisibility) {
-        self.0.lock().unwrap().visibility = visibility;
-    }
-
-    pub(crate) fn get_visibility(&self) -> LayerVisibility {
-        self.0.lock().unwrap().visibility.clone()
-    }
-
-    /// Summarize how likely this layer is to be used: its access time (if accessed), and its visibility hint.
-    pub(crate) fn atime_visibility(&self) -> (Option<SystemTime>, LayerVisibility) {
-        let state = self.0.lock().unwrap();
-
-        (
-            state
-                .for_eviction_policy
-                .last_accesses
-                .recent()
-                .map(|a| a.when),
-            state.visibility.clone(),
-        )
-    }
 }
 
 /// Get a layer descriptor from a layer.

pageserver/src/tenant/storage_layer/layer.rs

@@ -250,8 +250,6 @@ impl Layer {
                 LayerResidenceStatus::Resident,
                 LayerResidenceEventReason::LayerCreate,
             );
-            // Newly created layers are marked visible by default: the usual case is that they were created to be read.
-            access_stats.set_visibility(super::LayerVisibility::Visible);
 
             let local_path = local_layer_path(
                 conf,

pageserver/src/tenant/timeline.rs

@@ -30,7 +30,7 @@ use pageserver_api::{
         InMemoryLayerInfo, LayerMapInfo, LsnLease, TimelineState,
     },
     reltag::BlockNumber,
-    shard::{ShardCount, ShardIdentity, ShardNumber, TenantShardId},
+    shard::{ShardIdentity, ShardNumber, TenantShardId},
 };
 use rand::Rng;
 use serde_with::serde_as;
@@ -135,7 +135,7 @@ use self::layer_manager::LayerManager;
 use self::logical_size::LogicalSize;
 use self::walreceiver::{WalReceiver, WalReceiverConf};
 
-use super::{config::TenantConf, storage_layer::LayerVisibility};
+use super::config::TenantConf;
 use super::{debug_assert_current_span_has_tenant_and_timeline_id, AttachedTenantConf};
 use super::{remote_timeline_client::index::IndexPart, storage_layer::LayerFringe};
 use super::{remote_timeline_client::RemoteTimelineClient, storage_layer::ReadableLayer};
@@ -453,12 +453,12 @@ pub struct WalReceiverInfo {
 /// Garbage Collection.
 #[derive(Default)]
 pub(crate) struct GcInfo {
-    /// Record which parts of this timeline's history are still needed by children
+    /// Specific LSNs that are needed.
     ///
-    /// Optionally store each child's keyspace at their branch LSN: parts of the keyspace not covered here may be dropped during GC, as
-    /// the child will never read them.  For example, a child which has covered its whole keyspace with image layers
-    /// will put an empty keyspace here.  Children populate this: if it is None, presume the child may read any part of the keyspace.
-    pub(crate) retain_lsns: Vec<(Lsn, TimelineId, Option<KeySpace>)>,
+    /// Currently, this includes all points where child branches have
+    /// been forked off from. In the future, could also include
+    /// explicit user-defined snapshot points.
+    pub(crate) retain_lsns: Vec<Lsn>,
 
     /// The cutoff coordinates, which are combined by selecting the minimum.
     pub(crate) cutoffs: GcCutoffs,
@@ -474,23 +474,6 @@ impl GcInfo {
     pub(crate) fn min_cutoff(&self) -> Lsn {
         self.cutoffs.select_min()
     }
-
-    pub(super) fn insert_child(&mut self, child_id: TimelineId, child_lsn: Lsn) {
-        self.retain_lsns.push((child_lsn, child_id, None));
-        self.retain_lsns.sort_by_key(|i| i.0);
-    }
-
-    pub(super) fn remove_child(&mut self, child_id: TimelineId) {
-        self.retain_lsns.retain(|i| i.1 != child_id);
-    }
-
-    /// When the child re-calculates which parts of the keyspace it will read from the ancestor, it posts
-    /// and update to the parent using this function, to enable the parent to perhaps GC more layers.
-    pub(super) fn notify_child_keyspace(&mut self, child_id: TimelineId, key_space: KeySpace) {
-        if let Ok(idx) = self.retain_lsns.binary_search_by_key(&child_id, |i| i.1) {
-            self.retain_lsns.get_mut(idx).unwrap().2 = Some(key_space);
-        }
-    }
 }
 
 /// The `GcInfo` component describing which Lsns need to be retained.
@@ -1810,26 +1793,9 @@ impl Timeline {
         }
 
         match self.get_compaction_algorithm_settings().kind {
-            CompactionAlgorithm::Tiered => self.compact_tiered(cancel, ctx).await?,
-            CompactionAlgorithm::Legacy => self.compact_legacy(cancel, flags, ctx).await?,
+            CompactionAlgorithm::Tiered => self.compact_tiered(cancel, ctx).await,
+            CompactionAlgorithm::Legacy => self.compact_legacy(cancel, flags, ctx).await,
         }
-
-        if self.shard_identity.count >= ShardCount::new(2) {
-            // Limit the number of layer rewrites to the number of partitions: this means its
-            // runtime should be comparable to a full round of image layer creations, rather than
-            // being potentially much longer.
-            // TODO: make `partitioning` a sync lock: see comment in `repartition()` for why there's no
-            // real async use.
-            let rewrite_max = self.partitioning.try_lock().unwrap().0 .0.parts.len();
-
-            self.compact_shard_ancestors(rewrite_max, ctx).await?;
-        }
-
-        // TODO: be more selective: call this once at startup, and thereafter only when some branching changes or
-        // when image layer are generated.
-        self.update_layer_visibility(ctx).await?;
-
-        Ok(())
     }
 
     /// Mutate the timeline with a [`TimelineWriter`].
@@ -3001,17 +2967,6 @@ impl Timeline {
             .set((calculated_size, metrics_guard.calculation_result_saved()))
             .ok()
             .expect("only this task sets it");
-
-        // As a nice-to-have, calculate layer visibilties.  Otherwise this will
-        // be initialized on first compaction.  Doing it as early as possible
-        // enables code that depends on layer visibility (like uploading heatmaps)
-        // to execute earlier, rather than waiting for compaction.
-        match self.update_layer_visibility(&background_ctx).await {
-            Ok(_) | Err(CompactionError::ShuttingDown) => {}
-            Err(e) => {
-                tracing::warn!("Initial layer visibility calculation failed: {e}");
-            }
-        }
     }
 
     pub(crate) fn spawn_ondemand_logical_size_calculation(
@@ -3189,8 +3144,7 @@ impl Timeline {
     }
 
     /// The timeline heatmap is a hint to secondary locations from the primary location,
-    /// indicating which layers should be downloaded on the secondary to give it a warm
-    /// cache, that will enable it to take over as the attached location without degrading performance.
+    /// indicating which layers are currently on-disk on the primary.
     ///
     /// None is returned if the Timeline is in a state where uploading a heatmap
     /// doesn't make sense, such as shutting down or initializing.  The caller
@@ -3203,32 +3157,19 @@ impl Timeline {
 
         let guard = self.layers.read().await;
 
-        let mut resident_visible_layers = Vec::new();
-        let now = SystemTime::now();
-        for layer in guard.likely_resident_layers() {
-            let (atime, visibility) = layer.access_stats().atime_visibility();
+        let resident = guard.likely_resident_layers().map(|layer| {
+            let last_activity_ts = layer.access_stats().latest_activity_or_now();
 
-            match visibility {
-                LayerVisibility::Uninitialized => {
-                    // Refuse to generate a heatmap at all until layer visibilty is initialized
-                    return None;
-                }
-                LayerVisibility::Covered => {
-                    // This layer is covered: exclude it from the heatmap because a secondary
-                    // node is highly unlikely to need this layer in the event that it takes over as attached
-                }
-                LayerVisibility::Visible => resident_visible_layers.push(HeatMapLayer::new(
-                    layer.layer_desc().layer_name(),
-                    layer.metadata(),
-                    atime.unwrap_or(now),
-                )),
-            }
-        }
+            HeatMapLayer::new(
+                layer.layer_desc().layer_name(),
+                layer.metadata(),
+                last_activity_ts,
+            )
+        });
 
-        Some(HeatMapTimeline::new(
-            self.timeline_id,
-            resident_visible_layers,
-        ))
+        let layers = resident.collect();
+
+        Some(HeatMapTimeline::new(self.timeline_id, layers))
     }
 
     /// Returns true if the given lsn is or was an ancestor branchpoint.
@@ -5094,11 +5035,7 @@ impl Timeline {
 
             let horizon_cutoff = min(gc_info.cutoffs.horizon, self.get_disk_consistent_lsn());
             let pitr_cutoff = gc_info.cutoffs.pitr;
-            let retain_lsns = gc_info
-                .retain_lsns
-                .iter()
-                .map(|(lsn, _child_id, _)| *lsn)
-                .collect();
+            let retain_lsns = gc_info.retain_lsns.clone();
 
             // Gets the maximum LSN that holds the valid lease.
             //

pageserver/src/tenant/timeline/compaction.rs

@@ -19,14 +19,14 @@ use enumset::EnumSet;
 use fail::fail_point;
 use itertools::Itertools;
 use pageserver_api::keyspace::ShardedRange;
-use pageserver_api::shard::{ShardIdentity, TenantShardId};
+use pageserver_api::shard::{ShardCount, ShardIdentity, TenantShardId};
 use tokio_util::sync::CancellationToken;
 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, LayerVisibility, PersistentLayerDesc};
+use crate::tenant::storage_layer::{AsLayerDesc, PersistentLayerDesc};
 use crate::tenant::timeline::{drop_rlock, Hole, ImageLayerCreationOutcome};
 use crate::tenant::timeline::{DeltaLayerWriter, ImageLayerWriter};
 use crate::tenant::timeline::{Layer, ResidentLayer};
@@ -100,7 +100,7 @@ impl Timeline {
         // Define partitioning schema if needed
 
         // FIXME: the match should only cover repartitioning, not the next steps
-        match self
+        let partition_count = match self
             .repartition(
                 self.get_last_record_lsn(),
                 self.get_compaction_target_size(),
@@ -140,6 +140,7 @@ impl Timeline {
                     .await?;
 
                 self.upload_new_image_layers(image_layers)?;
+                partitioning.parts.len()
             }
             Err(err) => {
                 // no partitioning? This is normal, if the timeline was just created
@@ -151,9 +152,19 @@ impl Timeline {
                 if !self.cancel.is_cancelled() {
                     tracing::error!("could not compact, repartitioning keyspace failed: {err:?}");
                 }
+                1
             }
         };
 
+        if self.shard_identity.count >= ShardCount::new(2) {
+            // Limit the number of layer rewrites to the number of partitions: this means its
+            // runtime should be comparable to a full round of image layer creations, rather than
+            // being potentially much longer.
+            let rewrite_max = partition_count;
+
+            self.compact_shard_ancestors(rewrite_max, ctx).await?;
+        }
+
         Ok(())
     }
 
@@ -165,7 +176,7 @@ impl Timeline {
     ///
     /// Note: this phase may read and write many gigabytes of data: use rewrite_max to bound
     /// how much work it will try to do in each compaction pass.
-    pub(super) async fn compact_shard_ancestors(
+    async fn compact_shard_ancestors(
         self: &Arc<Self>,
         rewrite_max: usize,
         ctx: &RequestContext,
@@ -347,88 +358,6 @@ impl Timeline {
         Ok(())
     }
 
-    /// A post-compaction step to update the LayerVisibility of layers covered by image layers.  This
-    /// should also be called when new branches are created.
-    ///
-    /// Sweep through the layer map, identifying layers which are covered by image layers
-    /// such that they do not need to be available to service reads. The resulting LayerVisibility
-    /// result may be used as an input to eviction and secondary downloads to de-prioritize layers
-    /// that we know won't be needed for reads.
-    pub(super) async fn update_layer_visibility(
-        &self,
-        ctx: &RequestContext,
-    ) -> Result<(), CompactionError> {
-        // Start with a keyspace representing all the keys we need to read from the tip of the branch
-        let head_lsn = self.get_last_record_lsn();
-        let (mut head_keyspace, sparse_ks) = self.collect_keyspace(head_lsn, ctx).await?;
-
-        // Converting the sparse part of the keyspace into the dense keyspace is safe in this context
-        // because we will never iterate through the keys.
-        head_keyspace.merge(&sparse_ks.0);
-
-        // We will sweep through layers in reverse-LSN order.  We only do historic layers.  L0 deltas
-        // are implicitly visible, because LayerVisibility's default is Visible, and we never modify it here.
-        let layer_manager = self.layers.read().await;
-        let layer_map = layer_manager.layer_map();
-
-        let mut visible_size: u64 = 0;
-
-        // FIXME: we only get accurate keyspaces from children if they've already run update_layer_visibility themselves.  At startup all the timelines
-        // initialize this in arbitrary order (at the end of initial_logical_size_calculation).  We should coordinate these.  Perhaps at the very start
-        // of the tenant compaction task we should do all the timelines' layer visibility calculations in a leaf-first order?
-        let readable_points = {
-            let children = self.gc_info.read().unwrap().retain_lsns.clone();
-
-            let mut readable_points = Vec::with_capacity(children.len() + 1);
-            for (child_lsn, _child_timeline_id, child_keyspace) in &children {
-                let keyspace = match child_keyspace {
-                    Some(ks) => ks.clone(),
-                    None => {
-                        // The child has not posted information about which parts of the keyspace they depend on: presume they depend on all of it.
-                        let (mut keyspace, sparse_keyspace) =
-                            self.collect_keyspace(*child_lsn, ctx).await?;
-                        keyspace.merge(&sparse_keyspace.0);
-                        keyspace
-                    }
-                };
-                readable_points.push((*child_lsn, keyspace));
-            }
-            readable_points.push((head_lsn, head_keyspace));
-            readable_points
-        };
-
-        let (layer_visibility, shadow) = layer_map.get_visibility(readable_points);
-        for (layer_desc, visibility) in layer_visibility {
-            // FIXME: a more efficiency bulk zip() through the layers rather than NlogN getting each one
-            let layer = layer_manager.get_from_desc(&layer_desc);
-            if matches!(visibility, LayerVisibility::Visible) {
-                visible_size += layer.metadata().file_size;
-            }
-
-            layer.access_stats().set_visibility(visibility);
-        }
-
-        if let Some(ancestor) = &self.ancestor_timeline {
-            // Having calculated the readable keyspace after walking back through all this timeline's layers, the resulting keyspace is the remaining
-            // keys for which reads may still fall through to the parent branch.  Notify the parent branch of this, so that they may GC layers which
-            // do not overlap with this keyspace, and so that they may use this as an input to their own visibility updates.
-            ancestor
-                .gc_info
-                .write()
-                .unwrap()
-                .notify_child_keyspace(self.timeline_id, shadow);
-        }
-
-        // Also include in the visible size all the layers which we would never update visibility on
-        // TODO: getter that doesn't spuriously construct a Vec<>
-        for layer in layer_map.get_level0_deltas().unwrap() {
-            visible_size += layer.file_size;
-        }
-        self.metrics.visible_physical_size_gauge.set(visible_size);
-
-        Ok(())
-    }
-
     /// Collect a bunch of Level 0 layer files, and compact and reshuffle them as
     /// as Level 1 files.
     async fn compact_level0(

pageserver/src/tenant/timeline/delete.rs

@@ -148,14 +148,14 @@ async fn cleanup_remaining_timeline_fs_traces(
 /// For more context see comments in [`DeleteTimelineFlow::prepare`]
 async fn remove_timeline_from_tenant(
     tenant: &Tenant,
-    timeline: &Timeline,
+    timeline_id: TimelineId,
     _: &DeletionGuard, // using it as a witness
 ) -> anyhow::Result<()> {
     // Remove the timeline from the map.
     let mut timelines = tenant.timelines.lock().unwrap();
     let children_exist = timelines
         .iter()
-        .any(|(_, entry)| entry.get_ancestor_timeline_id() == Some(timeline.timeline_id));
+        .any(|(_, entry)| entry.get_ancestor_timeline_id() == Some(timeline_id));
     // XXX this can happen because `branch_timeline` doesn't check `TimelineState::Stopping`.
     // We already deleted the layer files, so it's probably best to panic.
     // (Ideally, above remove_dir_all is atomic so we don't see this timeline after a restart)
@@ -163,14 +163,8 @@ async fn remove_timeline_from_tenant(
         panic!("Timeline grew children while we removed layer files");
     }
 
-    // Unlink from parent
-    if let Some(ancestor) = timeline.get_ancestor_timeline() {
-        let mut ancestor_gc_info = ancestor.gc_info.write().unwrap();
-        ancestor_gc_info.remove_child(timeline.timeline_id);
-    }
-
     timelines
-        .remove(&timeline.timeline_id)
+        .remove(&timeline_id)
         .expect("timeline that we were deleting was concurrently removed from 'timelines' map");
 
     drop(timelines);
@@ -299,9 +293,6 @@ impl DeleteTimelineFlow {
         {
             let mut locked = tenant.timelines.lock().unwrap();
             locked.insert(timeline_id, Arc::clone(&timeline));
-
-            // Note that we do not insert this into the parent branch's GcInfo: the parent is not obliged to retain
-            // any data for child timelines being deleted.
         }
 
         guard.mark_in_progress()?;
@@ -422,7 +413,7 @@ impl DeleteTimelineFlow {
 
         pausable_failpoint!("in_progress_delete");
 
-        remove_timeline_from_tenant(tenant, timeline, &guard).await?;
+        remove_timeline_from_tenant(tenant, timeline.timeline_id, &guard).await?;
 
         *guard = Self::Finished;
 

pageserver/src/tenant/timeline/layer_manager.rs

@@ -255,14 +255,6 @@ impl LayerManager {
                 new_layer.layer_desc().lsn_range
             );
 
-            // Transfer visibilty hint from old to new layer, since the new layer covers the same key space.  This is not guaranteed to
-            // be accurate (as the new layer may cover a different subset of the key range), but is a sensible default, and prevents
-            // always marking rewritten layers as visible.
-            new_layer
-                .as_ref()
-                .access_stats()
-                .set_visibility(old_layer.access_stats().get_visibility());
-
             // Safety: we may never rewrite the same file in-place.  Callers are responsible
             // for ensuring that they only rewrite layers after something changes the path,
             // such as an increment in the generation number.

pageserver/src/walingest.rs

@@ -343,33 +343,7 @@ impl WalIngest {
                         xlog_checkpoint.oldestActiveXid,
                         self.checkpoint.oldestActiveXid
                     );
-
-                    // A shutdown checkpoint has `oldestActiveXid == InvalidTransactionid`,
-                    // because at shutdown, all in-progress transactions will implicitly
-                    // end. Postgres startup code knows that, and allows hot standby to start
-                    // immediately from a shutdown checkpoint.
-                    //
-                    // In Neon, Postgres hot standby startup always behaves as if starting from
-                    // an online checkpoint. It needs a valid `oldestActiveXid` value, so
-                    // instead of overwriting self.checkpoint.oldestActiveXid with
-                    // InvalidTransactionid from the checkpoint WAL record, update it to a
-                    // proper value, knowing that there are no in-progress transactions at this
-                    // point, except for prepared transactions.
-                    //
-                    // See also the neon code changes in the InitWalRecovery() function.
-                    if xlog_checkpoint.oldestActiveXid == pg_constants::INVALID_TRANSACTION_ID
-                        && info == pg_constants::XLOG_CHECKPOINT_SHUTDOWN
-                    {
-                        let mut oldest_active_xid = self.checkpoint.nextXid.value as u32;
-                        for xid in modification.tline.list_twophase_files(lsn, ctx).await? {
-                            if (xid.wrapping_sub(oldest_active_xid) as i32) < 0 {
-                                oldest_active_xid = xid;
-                            }
-                        }
-                        self.checkpoint.oldestActiveXid = oldest_active_xid;
-                    } else {
-                        self.checkpoint.oldestActiveXid = xlog_checkpoint.oldestActiveXid;
-                    }
+                    self.checkpoint.oldestActiveXid = xlog_checkpoint.oldestActiveXid;
 
                     // Write a new checkpoint key-value pair on every checkpoint record, even
                     // if nothing really changed. Not strictly required, but it seems nice to
@@ -401,7 +375,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;
                 }
             }
             pg_constants::RM_REPLORIGIN_ID => {
@@ -1304,10 +1277,13 @@ impl WalIngest {
             xlrec.pageno, xlrec.oldest_xid, xlrec.oldest_xid_db
         );
 
-        // In Postgres, oldestXid and oldestXidDB are updated in memory when the CLOG is
-        // truncated, but a checkpoint record with the updated values isn't written until
-        // later. In Neon, a server can start at any LSN, not just on a checkpoint record,
-        // so we keep the oldestXid and oldestXidDB up-to-date.
+        // Here we treat oldestXid and oldestXidDB
+        // differently from postgres redo routines.
+        // In postgres checkpoint.oldestXid lags behind xlrec.oldest_xid
+        // until checkpoint happens and updates the value.
+        // Here we can use the most recent value.
+        // It's just an optimization, though and can be deleted.
+        // TODO Figure out if there will be any issues with replica.
         self.checkpoint.oldestXid = xlrec.oldest_xid;
         self.checkpoint.oldestXidDB = xlrec.oldest_xid_db;
         self.checkpoint_modified = true;

pgxn/neon/neon.c

@@ -12,8 +12,6 @@
 #include "fmgr.h"
 
 #include "miscadmin.h"
-#include "access/subtrans.h"
-#include "access/twophase.h"
 #include "access/xact.h"
 #include "access/xlog.h"
 #include "storage/buf_internals.h"
@@ -24,12 +22,10 @@
 #include "replication/logical.h"
 #include "replication/slot.h"
 #include "replication/walsender.h"
-#include "storage/proc.h"
 #include "storage/procsignal.h"
 #include "tcop/tcopprot.h"
 #include "funcapi.h"
 #include "access/htup_details.h"
-#include "utils/builtins.h"
 #include "utils/pg_lsn.h"
 #include "utils/guc.h"
 #include "utils/wait_event.h"
@@ -270,293 +266,6 @@ LogicalSlotsMonitorMain(Datum main_arg)
 	}
 }
 
-/*
- * XXX: These private to procarray.c, but we need them here.
- */
-#define PROCARRAY_MAXPROCS	(MaxBackends + max_prepared_xacts)
-#define TOTAL_MAX_CACHED_SUBXIDS \
-	((PGPROC_MAX_CACHED_SUBXIDS + 1) * PROCARRAY_MAXPROCS)
-
-/*
- * Restore running-xact information by scanning the CLOG at startup.
- *
- * In PostgreSQL, a standby always has to wait for a running-xacts WAL record
- * to arrive before it can start accepting queries. Furthermore, if there are
- * transactions with too many subxids (> 64) open to fit in the in-memory
- * subxids cache, the running-xacts record will be marked as "suboverflowed",
- * and the standby will need to also wait for the currently in-progress
- * transactions to finish.
- *
- * That's not great in PostgreSQL, because a hot standby does not necessary
- * open up for queries immediately as you might expect. But it's worse in
- * Neon: A standby in Neon doesn't need to start WAL replay from a checkpoint
- * record; it can start at any LSN. Postgres arranges things so that there is
- * a running-xacts record soon after every checkpoint record, but when you
- * start from an arbitrary LSN, that doesn't help. If the primary is idle, or
- * not running at all, it might never write a new running-xacts record,
- * leaving the replica in a limbo where it can never start accepting queries.
- *
- * To mitigate that, we have an additional mechanism to find the running-xacts
- * information: we scan the CLOG, making note of any XIDs not marked as
- * committed or aborted. They are added to the Postgres known-assigned XIDs
- * array by calling ProcArrayApplyRecoveryInfo() in the caller of this
- * function.
- *
- * There is one big limitation with that mechanism: The size of the
- * known-assigned XIDs is limited, so if there are a lot of in-progress XIDs,
- * we have to give up. Furthermore, we don't know how many of the in-progress
- * XIDs are subtransactions, and if we use up all the space in the
- * known-assigned XIDs array for subtransactions, we might run out of space in
- * the array later during WAL replay, causing the replica to shut down with
- * "ERROR: too many KnownAssignedXids". The safe # of XIDs that we can add to
- * the known-assigned array without risking that error later is very low,
- * merely PGPROC_MAX_CACHED_SUBXIDS == 64, so we take our chances and use up
- * to half of the known-assigned XIDs array for the subtransactions, even
- * though that risks getting the error later.
- *
- * Note: It's OK if the recovered list of XIDs includes some transactions that
- * have crashed in the primary, and hence will never commit. They will be seen
- * as in-progress, until we see a new next running-acts record with an
- * oldestActiveXid that invalidates them. That's how the known-assigned XIDs
- * array always works.
- *
- * If scraping the CLOG doesn't succeed for some reason, like the subxid
- * overflow, Postgres will fall back to waiting for a running-xacts record
- * like usual.
- *
- * Returns true if a complete list of in-progress XIDs was scraped.
- */
-static bool
-RestoreRunningXactsFromClog(CheckPoint *checkpoint, TransactionId **xids, int *nxids)
-{
-	TransactionId from;
-	TransactionId till;
-	int			max_xcnt;
-	TransactionId *prepared_xids = NULL;
-	int			n_prepared_xids;
-	TransactionId *restored_xids = NULL;
-	int			n_restored_xids;
-	int			next_prepared_idx;
-
-	Assert(*xids == NULL);
-
-	/*
-	 * If the checkpoint doesn't have a valid oldestActiveXid, bail out. We
-	 * don't know where to start the scan.
-	 *
-	 * This shouldn't happen, because the pageserver always maintains a valid
-	 * oldestActiveXid nowadays. Except when starting at an old point in time
-	 * that was ingested before the pageserver was taught to do that.
-	 */
-	if (!TransactionIdIsValid(checkpoint->oldestActiveXid))
-	{
-		elog(LOG, "cannot restore running-xacts from CLOG because oldestActiveXid is not set");
-		goto fail;
-	}
-
-	/*
-	 * We will scan the CLOG starting from the oldest active XID.
-	 *
-	 * In some corner cases, the oldestActiveXid from the last checkpoint
-	 * might already have been truncated from the CLOG. That is,
-	 * oldestActiveXid might be older than oldestXid. That's possible because
-	 * oldestActiveXid is only updated at checkpoints. After the last
-	 * checkpoint, the oldest transaction might have committed, and the CLOG
-	 * might also have been already truncated. So if oldestActiveXid is older
-	 * than oldestXid, start at oldestXid instead. (Otherwise we'd try to
-	 * access CLOG segments that have already been truncated away.)
-	 */
-	from = TransactionIdPrecedes(checkpoint->oldestXid, checkpoint->oldestActiveXid)
-		? checkpoint->oldestActiveXid : checkpoint->oldestXid;
-	till = XidFromFullTransactionId(checkpoint->nextXid);
-
-	/*
-	 * To avoid "too many KnownAssignedXids" error later during replay, we
-	 * limit number of collected 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.
-	 *
-	 * The size of the KnownAssignedXids array is TOTAL_MAX_CACHED_SUBXIDS,
-	 * which is (PGPROC_MAX_CACHED_SUBXIDS + 1) * PROCARRAY_MAXPROCS). In
-	 * PostgreSQL, that's always enough because the primary will always write
-	 * an XLOG_XACT_ASSIGNMENT record if a transaction has more than
-	 * PGPROC_MAX_CACHED_SUBXIDS subtransactions. Seeing that record allows
-	 * the standby to mark the XIDs in pg_subtrans and removing them from the
-	 * KnowingAssignedXids array.
-	 *
-	 * Here, we don't know which XIDs belong to subtransactions that have
-	 * already been WAL-logged with an XLOG_XACT_ASSIGNMENT record. If we
-	 * wanted to be totally safe and avoid the possibility of getting a "too
-	 * many KnownAssignedXids" error later, we would have to limit ourselves
-	 * to PGPROC_MAX_CACHED_SUBXIDS, which is not much. And that includes top
-	 * transaction IDs too, because we cannot distinguish between top
-	 * transaction IDs and subtransactions here.
-	 *
-	 * Somewhat arbitrarily, we use up to half of KnownAssignedXids. That
-	 * strikes a sensible balance between being useful, and risking a "too
-	 * many KnownAssignedXids" error later.
-	 */
-	max_xcnt = TOTAL_MAX_CACHED_SUBXIDS / 2;
-
-	/*
-	 * Collect XIDs of prepared transactions in an array. This includes only
-	 * their top-level XIDs. We assume that StandbyRecoverPreparedTransactions
-	 * has already been called, so we can find all the sub-transactions in
-	 * pg_subtrans.
-	 */
-	PrescanPreparedTransactions(&prepared_xids, &n_prepared_xids);
-	qsort(prepared_xids, n_prepared_xids, sizeof(TransactionId), xidLogicalComparator);
-
-	/*
-	 * Scan the CLOG, collecting in-progress XIDs into 'restored_xids'.
-	 */
-	elog(DEBUG1, "scanning CLOG between %u and %u for in-progress XIDs", from, till);
-	restored_xids = (TransactionId *) palloc(max_xcnt * sizeof(TransactionId));
-	n_restored_xids = 0;
-	next_prepared_idx = 0;
-	for (TransactionId xid = from; xid != till;)
-	{
-		XLogRecPtr	xidlsn;
-		XidStatus	xidstatus;
-
-		xidstatus = TransactionIdGetStatus(xid, &xidlsn);
-
-		/*
-		 * "Merge" the prepared transactions into the restored_xids array as
-		 * we go.  The prepared transactions array is sorted. This is mostly
-		 * a sanity check to ensure that all the prpeared transactions are
-		 * seen as in-progress. (There is a check after the loop that we didn't
-		 * miss any.)
-		 */
-		if (next_prepared_idx < n_prepared_xids && xid == prepared_xids[next_prepared_idx])
-		{
-			/*
-			 * This is a top-level transaction ID of a prepared transaction.
-			 * Include it in the array.
-			 */
-
-			/* sanity check */
-			if (xidstatus != TRANSACTION_STATUS_IN_PROGRESS)
-			{
-				elog(LOG, "prepared transaction %u has unexpected status %X, cannot restore running-xacts from CLOG",
-					 xid, xidstatus);
-				Assert(false);
-				goto fail;
-			}
-
-			elog(DEBUG1, "XID %u: was next prepared xact (%d / %d)", xid, next_prepared_idx, n_prepared_xids);
-			next_prepared_idx++;
-		}
-		else if (xidstatus == TRANSACTION_STATUS_COMMITTED)
-		{
-			elog(DEBUG1, "XID %u: was committed", xid);
-			goto skip;
-		}
-		else if (xidstatus == TRANSACTION_STATUS_ABORTED)
-		{
-			elog(DEBUG1, "XID %u: was aborted", xid);
-			goto skip;
-		}
-		else if (xidstatus == TRANSACTION_STATUS_IN_PROGRESS)
-		{
-			/*
-			 * In-progress transactions are included in the array.
-			 *
-			 * Except subtransactions of the prepared transactions. They are
-			 * already set in pg_subtrans, and hence don't need to be tracked
-			 * in the known-assigned XIDs array.
-			 */
-			if (n_prepared_xids > 0)
-			{
-				TransactionId parent = SubTransGetParent(xid);
-
-				if (TransactionIdIsValid(parent))
-				{
-					/*
-					 * This is a subtransaction belonging to a prepared
-					 * transaction.
-					 *
-					 * Sanity check that it is in the prepared XIDs array. It
-					 * should be, because StandbyRecoverPreparedTransactions
-					 * populated pg_subtrans, and no other XID should be set
-					 * in it yet. (This also relies on the fact that
-					 * StandbyRecoverPreparedTransactions sets the parent of
-					 * each subxid to point directly to the top-level XID,
-					 * rather than restoring the original subtransaction
-					 * hierarchy.)
-					 */
-					if (bsearch(&parent, prepared_xids, next_prepared_idx,
-								sizeof(TransactionId), xidLogicalComparator) == NULL)
-					{
-						elog(LOG, "sub-XID %u has unexpected parent %u, cannot restore running-xacts from CLOG",
-							 xid, parent);
-						Assert(false);
-						goto fail;
-					}
-					elog(DEBUG1, "XID %u: was a subtransaction of prepared xid %u", xid, parent);
-					goto skip;
-				}
-			}
-
-			/* include it in the array */
-			elog(DEBUG1, "XID %u: is in progress", xid);
-		}
-		else
-		{
-			/*
-			 * SUB_COMMITTED is a transient state used at commit. We don't
-			 * expect to see that here.
-			 */
-			elog(LOG, "XID %u has unexpected status %X in pg_xact, cannot restore running-xacts from CLOG",
-				 xid, xidstatus);
-			Assert(false);
-			goto fail;
-		}
-
-		if (n_restored_xids >= max_xcnt)
-		{
-			/*
-			 * Overflowed. We won't be able to install the RunningTransactions
-			 * snapshot.
-			 */
-			elog(LOG, "too many running xacts to restore from the CLOG; oldestXid=%u oldestActiveXid=%u nextXid %u",
-				 checkpoint->oldestXid, checkpoint->oldestActiveXid,
-				 XidFromFullTransactionId(checkpoint->nextXid));
-			goto fail;
-		}
-
-		restored_xids[n_restored_xids++] = xid;
-
-	skip:
-		TransactionIdAdvance(xid);
-		continue;
-	}
-
-	/* sanity check */
-	if (next_prepared_idx != n_prepared_xids)
-	{
-		elog(LOG, "prepared transaction ID %u was not visited in the CLOG scan, cannot restore running-xacts from CLOG",
-			 prepared_xids[next_prepared_idx]);
-		Assert(false);
-		goto fail;
-	}
-
-	elog(LOG, "restored %d running xacts by scanning the CLOG; oldestXid=%u oldestActiveXid=%u nextXid %u",
-		 n_restored_xids, checkpoint->oldestXid, checkpoint->oldestActiveXid, XidFromFullTransactionId(checkpoint->nextXid));
-	*nxids = n_restored_xids;
-	*xids = restored_xids;
-	return true;
-
- fail:
-	*nxids = 0;
-	*xids = NULL;
-	if (restored_xids)
-		pfree(restored_xids);
-	if (prepared_xids)
-		pfree(prepared_xids);
-	return false;
-}
-
 void
 _PG_init(void)
 {
@@ -579,8 +288,6 @@ _PG_init(void)
 
 	pg_init_extension_server();
 
-	restore_running_xacts_callback = RestoreRunningXactsFromClog;
-
 	/*
 	 * Important: This must happen after other parts of the extension are
 	 * loaded, otherwise any settings to GUCs that were set before the

poetry.lock

@@ -1,4 +1,4 @@
-# This file is automatically @generated by Poetry 1.8.2 and should not be changed by hand.
+# This file is automatically @generated by Poetry 1.8.3 and should not be changed by hand.
 
 [[package]]
 name = "aiohttp"
@@ -734,13 +734,13 @@ typing-extensions = ">=4.1.0"
 
 [[package]]
 name = "certifi"
-version = "2023.7.22"
+version = "2024.7.4"
 description = "Python package for providing Mozilla's CA Bundle."
 optional = false
 python-versions = ">=3.6"
 files = [
-    {file = "certifi-2023.7.22-py3-none-any.whl", hash = "sha256:92d6037539857d8206b8f6ae472e8b77db8058fec5937a1ef3f54304089edbb9"},
-    {file = "certifi-2023.7.22.tar.gz", hash = "sha256:539cc1d13202e33ca466e88b2807e29f4c13049d6d87031a3c110744495cb082"},
+    {file = "certifi-2024.7.4-py3-none-any.whl", hash = "sha256:c198e21b1289c2ab85ee4e67bb4b4ef3ead0892059901a8d5b622f24a1101e90"},
+    {file = "certifi-2024.7.4.tar.gz", hash = "sha256:5a1e7645bc0ec61a09e26c36f6106dd4cf40c6db3a1fb6352b0244e7fb057c7b"},
 ]
 
 [[package]]

test_runner/fixtures/neon_fixtures.py

@@ -3542,6 +3542,7 @@ class Endpoint(PgProtocol, LogUtils):
     ):
         super().__init__(host="localhost", port=pg_port, user="cloud_admin", dbname="postgres")
         self.env = env
+        self.running = False
         self.branch_name: Optional[str] = None  # dubious
         self.endpoint_id: Optional[str] = None  # dubious, see asserts below
         self.pgdata_dir: Optional[str] = None  # Path to computenode PGDATA
@@ -3915,9 +3916,7 @@ class EndpointFactory:
 
         return self
 
-    def new_replica(
-        self, origin: Endpoint, endpoint_id: str, config_lines: Optional[List[str]] = None
-    ):
+    def new_replica(self, origin: Endpoint, endpoint_id: str, config_lines: Optional[List[str]]):
         branch_name = origin.branch_name
         assert origin in self.endpoints
         assert branch_name is not None

test_runner/fixtures/pageserver/utils.py

@@ -198,7 +198,7 @@ def wait_for_last_record_lsn(
     lsn: Lsn,
 ) -> Lsn:
     """waits for pageserver to catch up to a certain lsn, returns the last observed lsn."""
-    for i in range(1000):
+    for i in range(100):
         current_lsn = last_record_lsn(pageserver_http, tenant, timeline)
         if current_lsn >= lsn:
             return current_lsn

test_runner/regress/test_replica_start.py

@@ -1,646 +0,0 @@
-"""
-In PostgreSQL, a standby always has to wait for a running-xacts WAL record to
-arrive before it can start accepting queries. Furthermore, if there are
-transactions with too many subxids (> 64) open to fit in the in-memory subxids
-cache, the running-xacts record will be marked as "suboverflowed", and the
-standby will need to also wait for the currently in-progress transactions to
-finish.
-
-In Neon, we have an additional mechanism that scans the CLOG at server startup
-to determine the list of running transactions, so that the standby can start up
-immediately without waiting for the running-xacts record, but that mechanism
-only works if the # of active (sub-)transactions is reasonably small. Otherwise
-it falls back to waiting. Furthermore, it's somewhat optimistic in using up the
-known-assigned XIDs array: if too many transactions with subxids are started in
-the primary later, the replay in the replica will crash with "too many
-KnownAssignedXids" error.
-
-This module contains tests for those various cases at standby startup: starting
-from shutdown checkpoint, using the CLOG scanning mechanism, waiting for
-running-xacts record and for in-progress transactions to finish etc.
-"""
-
-import threading
-from contextlib import closing
-
-import psycopg2
-import pytest
-from fixtures.log_helper import log
-from fixtures.neon_fixtures import NeonEnv, wait_for_last_flush_lsn, wait_replica_caughtup
-from fixtures.pg_version import PgVersion
-from fixtures.utils import query_scalar, wait_until
-
-CREATE_SUBXACTS_FUNC = """
-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: %', sqlerrm;
-      end;
-   end loop;
-end; $$ language plpgsql
-"""
-
-
-def test_replica_start_scan_clog(neon_simple_env: NeonEnv):
-    """
-    Test the CLOG-scanning mechanism at hot standby startup. There is one
-    transaction active in the primary when the standby is started. The primary
-    is killed before it has a chance to write a running-xacts record. The
-    CLOG-scanning at neon startup allows the standby to start up anyway.
-
-    See the module docstring for background.
-    """
-
-    # Initialize the primary, a test table, and a helper function to create lots
-    # of subtransactions.
-    env = neon_simple_env
-    primary = env.endpoints.create_start(branch_name="main", endpoint_id="primary")
-    primary_conn = primary.connect()
-    primary_cur = primary_conn.cursor()
-    primary_cur.execute("CREATE EXTENSION neon_test_utils")
-    primary_cur.execute("create table t(pk serial primary key, payload integer)")
-    primary_cur.execute(CREATE_SUBXACTS_FUNC)
-    primary_cur.execute("select pg_switch_wal()")
-
-    # Start a transaction in the primary. Leave the transaction open.
-    #
-    # The transaction has some subtransactions, but not too many to cause the
-    # CLOG-scanning mechanism to give up.
-    primary_cur.execute("begin")
-    primary_cur.execute("select create_subxacts(50)")
-
-    # Wait for the WAL to be flushed, but then immediately kill the primary,
-    # before it has a chance to generate a running-xacts record.
-    primary_cur.execute("select neon_xlogflush()")
-    wait_for_last_flush_lsn(env, primary, env.initial_tenant, env.initial_timeline)
-    primary.stop(mode="immediate")
-
-    # Create a replica. It should start up normally, thanks to the CLOG-scanning
-    # mechanism.
-    secondary = env.endpoints.new_replica_start(origin=primary, endpoint_id="secondary")
-
-    # The transaction did not commit, so it should not be visible in the secondary
-    secondary_conn = secondary.connect()
-    secondary_cur = secondary_conn.cursor()
-    secondary_cur.execute("select count(*) from t")
-    assert secondary_cur.fetchone() == (0,)
-
-
-def test_replica_start_scan_clog_crashed_xids(neon_simple_env: NeonEnv):
-    """
-    Test the CLOG-scanning mechanism at hot standby startup, after
-    leaving behind crashed transactions.
-
-    See the module docstring for background.
-    """
-
-    # Initialize the primary, a test table, and a helper function to create lots
-    # of subtransactions.
-    env = neon_simple_env
-    primary = env.endpoints.create_start(branch_name="main", endpoint_id="primary")
-    primary_conn = primary.connect()
-    primary_cur = primary_conn.cursor()
-    primary_cur.execute("create table t(pk serial primary key, payload integer)")
-    primary_cur.execute(CREATE_SUBXACTS_FUNC)
-    primary_cur.execute("select pg_switch_wal()")
-
-    # Consume a lot of XIDs, then kill Postgres without giving it a
-    # chance to write abort records for them.
-    primary_cur.execute("begin")
-    primary_cur.execute("select create_subxacts(100000)")
-    primary.stop(mode="immediate")
-
-    # Restart the primary. Do some light work, and shut it down cleanly
-    primary.start()
-    primary_conn = primary.connect()
-    primary_cur = primary_conn.cursor()
-    primary_cur.execute("insert into t (payload) values (0)")
-    primary.stop(mode="fast")
-
-    # Create a replica. It should start up normally, thanks to the CLOG-scanning
-    # mechanism. (Restarting the primary writes a checkpoint and/or running-xacts
-    # record, which allows the standby to know that the crashed XIDs are aborted)
-    secondary = env.endpoints.new_replica_start(origin=primary, endpoint_id="secondary")
-
-    secondary_conn = secondary.connect()
-    secondary_cur = secondary_conn.cursor()
-    secondary_cur.execute("select count(*) from t")
-    assert secondary_cur.fetchone() == (1,)
-
-
-def test_replica_start_at_running_xacts(neon_simple_env: NeonEnv, pg_version):
-    """
-    Test that starting a replica works right after the primary has
-    created a running-xacts record. This may seem like a trivial case,
-    but during development, we had a bug that was triggered by having
-    oldestActiveXid == nextXid. Starting right after a running-xacts
-    record is one way to test that case.
-
-    See the module docstring for background.
-    """
-    env = neon_simple_env
-
-    if env.pg_version == PgVersion.V14 or env.pg_version == PgVersion.V15:
-        pytest.skip("pg_log_standby_snapshot() function is available only in PG16")
-
-    primary = env.endpoints.create_start(branch_name="main", endpoint_id="primary")
-    primary_conn = primary.connect()
-    primary_cur = primary_conn.cursor()
-
-    primary_cur.execute("CREATE EXTENSION neon_test_utils")
-    primary_cur.execute("select pg_log_standby_snapshot()")
-    primary_cur.execute("select neon_xlogflush()")
-    wait_for_last_flush_lsn(env, primary, env.initial_tenant, env.initial_timeline)
-
-    secondary = env.endpoints.new_replica_start(origin=primary, endpoint_id="secondary")
-
-    secondary_conn = secondary.connect()
-    secondary_cur = secondary_conn.cursor()
-    secondary_cur.execute("select 123")
-    assert secondary_cur.fetchone() == (123,)
-
-
-def test_replica_start_wait_subxids_finish(neon_simple_env: NeonEnv):
-    """
-    Test replica startup when there are a lot of (sub)transactions active in the
-    primary. That's too many for the CLOG-scanning mechanism to handle, so the
-    replica has to wait for the large transaction to finish before it starts to
-    accept queries.
-
-    After replica startup, test MVCC with transactions that were in-progress
-    when the replica was started.
-
-    See the module docstring for background.
-    """
-
-    # Initialize the primary, a test table, and a helper function to create
-    # lots of subtransactions.
-    env = neon_simple_env
-    primary = env.endpoints.create_start(branch_name="main", endpoint_id="primary")
-    primary_conn = primary.connect()
-    primary_cur = primary_conn.cursor()
-    primary_cur.execute("create table t(pk serial primary key, payload integer)")
-    primary_cur.execute(CREATE_SUBXACTS_FUNC)
-
-    # Start a transaction with 100000 subtransactions, and leave it open. That's
-    # too many to fit in the "known-assigned XIDs array" in the replica, and
-    # also too many to fit in the subxid caches so the running-xacts record will
-    # also overflow.
-    primary_cur.execute("begin")
-    primary_cur.execute("select create_subxacts(100000)")
-
-    # Start another, smaller transaction in the primary. We'll come back to this
-    # later.
-    primary_conn2 = primary.connect()
-    primary_cur2 = primary_conn2.cursor()
-    primary_cur2.execute("begin")
-    primary_cur2.execute("insert into t (payload) values (0)")
-
-    # Create a replica. but before that, wait for the wal to be flushed to
-    # safekeepers, so that the replica is started at a point where the large
-    # transaction is already active. (The whole transaction might not be flushed
-    # yet, but that's OK.)
-    #
-    # Start it in a separate thread, so that we can do other stuff while it's
-    # blocked waiting for the startup to finish.
-    wait_for_last_flush_lsn(env, primary, env.initial_tenant, env.initial_timeline)
-    secondary = env.endpoints.new_replica(origin=primary, endpoint_id="secondary")
-    start_secondary_thread = threading.Thread(target=secondary.start)
-    start_secondary_thread.start()
-
-    # Verify that the replica has otherwise started up, but cannot start
-    # accepting queries yet.
-    log.info("Waiting 5 s to verify that the secondary does not start")
-    start_secondary_thread.join(5)
-    assert secondary.log_contains("consistent recovery state reached")
-    assert secondary.log_contains("started streaming WAL from primary")
-    # The "redo starts" message is printed when the first WAL record is
-    # received. It might or might not be present in the log depending on how
-    # far exactly the WAL was flushed when the replica was started, and whether
-    # background activity caused any more WAL records to be flushed on the
-    # primary afterwards.
-    #
-    # assert secondary.log_contains("redo # starts")
-
-    # should not be open for connections yet
-    assert start_secondary_thread.is_alive()
-    assert not secondary.is_running()
-    assert not secondary.log_contains("database system is ready to accept read-only connections")
-
-    # Commit the large transaction in the primary.
-    #
-    # Within the next 15 s, the primary should write a new running-xacts record
-    # to the WAL which shows the transaction as completed. Once the replica
-    # replays that record, it will start accepting queries.
-    primary_cur.execute("commit")
-    start_secondary_thread.join()
-
-    # Verify that the large transaction is correctly visible in the secondary
-    # (but not the second, small transaction, which is still in-progress!)
-    secondary_conn = secondary.connect()
-    secondary_cur = secondary_conn.cursor()
-    secondary_cur.execute("select count(*) from t")
-    assert secondary_cur.fetchone() == (100000,)
-
-    # Perform some more MVCC testing using the second transaction that was
-    # started in the primary before the replica was created
-    primary_cur2.execute("select create_subxacts(10000)")
-
-    # The second transaction still hasn't committed
-    wait_replica_caughtup(primary, secondary)
-    secondary_cur.execute("BEGIN ISOLATION LEVEL REPEATABLE READ")
-    secondary_cur.execute("select count(*) from t")
-    assert secondary_cur.fetchone() == (100000,)
-
-    # Commit the second transaction in the primary
-    primary_cur2.execute("commit")
-
-    # Should still be invisible to the old snapshot
-    wait_replica_caughtup(primary, secondary)
-    secondary_cur.execute("select count(*) from t")
-    assert secondary_cur.fetchone() == (100000,)
-
-    # Commit the REPEATABLE READ transaction in the replica. Both
-    # primary transactions should now be visible to a new snapshot.
-    secondary_cur.execute("commit")
-    secondary_cur.execute("select count(*) from t")
-    assert secondary_cur.fetchone() == (110001,)
-
-
-def test_replica_too_many_known_assigned_xids(neon_simple_env: NeonEnv):
-    """
-    The CLOG-scanning mechanism fills the known-assigned XIDs array
-    optimistically at standby startup, betting that it can still fit
-    upcoming transactions replayed later from the WAL in the
-    array. This test tests what happens when that bet fails and the
-    known-assigned XID array fills up after the standby has already
-    been started. The WAL redo will fail with an error:
-
-    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
-
-    which causes the standby to shut down.
-
-    See the module docstring for background.
-    """
-
-    # Initialize the primary, a test table, and a helper function to create lots
-    # of subtransactions.
-    env = neon_simple_env
-    primary = env.endpoints.create_start(branch_name="main", endpoint_id="primary")
-    primary_conn = primary.connect()
-    primary_cur = primary_conn.cursor()
-    primary_cur.execute("CREATE EXTENSION neon_test_utils")
-    primary_cur.execute("create table t(pk serial primary key, payload integer)")
-    primary_cur.execute(CREATE_SUBXACTS_FUNC)
-
-    # Determine how many connections we can use
-    primary_cur.execute("show max_connections")
-    max_connections = int(primary_cur.fetchall()[0][0])
-    primary_cur.execute("show superuser_reserved_connections")
-    superuser_reserved_connections = int(primary_cur.fetchall()[0][0])
-    n_connections = max_connections - superuser_reserved_connections
-    n_subxids = 200
-
-    # Start one top transaction in primary, with lots of subtransactions. This
-    # uses up much of 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(f"select create_subxacts({max_connections} * 30)")
-
-    with closing(primary.connect()) as small_p_conn:
-        with small_p_conn.cursor() as small_p_cur:
-            small_p_cur.execute("select create_subxacts(1)")
-
-    # Create a replica at this LSN
-    primary_cur.execute("select neon_xlogflush()")
-    wait_for_last_flush_lsn(env, primary, env.initial_tenant, env.initial_timeline)
-    secondary = env.endpoints.new_replica_start(origin=primary, endpoint_id="secondary")
-    secondary_conn = secondary.connect()
-    secondary_cur = secondary_conn.cursor()
-
-    # The transaction in primary has not committed yet.
-    wait_replica_caughtup(primary, secondary)
-    secondary_cur.execute("select count(*) from t")
-    assert secondary_cur.fetchone() == (1,)
-
-    # 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 to
-    # 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_curs = []
-    for _ in range(0, n_connections):
-        p_cur = primary.connect().cursor()
-        p_cur.execute("begin")
-        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)")
-
-    # Commit all the transactions in the primary
-    for i in range(0, n_connections):
-        p_curs[i].execute("commit")
-    large_p_cur.execute("commit")
-
-    # Wait until the replica crashes with "too many KnownAssignedXids" error.
-    def check_replica_crashed():
-        try:
-            secondary.connect()
-        except psycopg2.Error:
-            # Once the connection fails, return success
-            return None
-        raise RuntimeError("connection succeeded")
-
-    wait_until(20, 0.5, check_replica_crashed)
-    assert secondary.log_contains("too many KnownAssignedXids")
-
-    # Replica is crashed, so ignore stop result
-    secondary.check_stop_result = False
-
-
-def test_replica_start_repro_visibility_bug(neon_simple_env: NeonEnv):
-    """
-    Before PR #7288, a hot standby in neon incorrectly started up
-    immediately, before it had received a running-xacts record. That
-    led to visibility bugs if there were active transactions in the
-    primary. This test reproduces the incorrect query results and
-    incorrectly set hint bits, before that was fixed.
-    """
-    env = neon_simple_env
-
-    primary = env.endpoints.create_start(branch_name="main", endpoint_id="primary")
-    p_cur = primary.connect().cursor()
-
-    p_cur.execute("begin")
-    p_cur.execute("create table t(pk integer primary key, payload integer)")
-    p_cur.execute("insert into t values (generate_series(1,100000), 0)")
-
-    secondary = env.endpoints.new_replica_start(origin=primary, endpoint_id="secondary")
-    wait_replica_caughtup(primary, secondary)
-    s_cur = secondary.connect().cursor()
-
-    # Set hint bits for pg_class tuples. If primary's transaction is
-    # not marked as in-progress in MVCC snapshot, then XMIN_INVALID
-    # hint bit will be set for table's 't' tuple, making it invisible
-    # even after the commit record is replayed later.
-    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)
-
-
-@pytest.mark.parametrize("shutdown", [True, False])
-def test_replica_start_with_prepared_xacts(neon_simple_env: NeonEnv, shutdown: bool):
-    """
-    Test the CLOG-scanning mechanism at hot standby startup in the presence of
-    prepared transactions.
-
-    This test is run in two variants: one where the primary server is shut down
-    before starting the secondary, or not.
-    """
-
-    # Initialize the primary, a test table, and a helper function to create lots
-    # of subtransactions.
-    env = neon_simple_env
-    primary = env.endpoints.create_start(
-        branch_name="main", endpoint_id="primary", config_lines=["max_prepared_transactions=5"]
-    )
-    primary_conn = primary.connect()
-    primary_cur = primary_conn.cursor()
-    primary_cur.execute("CREATE EXTENSION neon_test_utils")
-    primary_cur.execute("create table t(pk serial primary key, payload integer)")
-    primary_cur.execute("create table t1(pk integer primary key)")
-    primary_cur.execute("create table t2(pk integer primary key)")
-    primary_cur.execute(CREATE_SUBXACTS_FUNC)
-
-    # Prepare a transaction for two-phase commit
-    primary_cur.execute("begin")
-    primary_cur.execute("insert into t1 values (1)")
-    primary_cur.execute("prepare transaction 't1'")
-
-    # Prepare another transaction for two-phase commit, with a subtransaction
-    primary_cur.execute("begin")
-    primary_cur.execute("insert into t2 values (2)")
-    primary_cur.execute("savepoint sp")
-    primary_cur.execute("insert into t2 values (3)")
-    primary_cur.execute("prepare transaction 't2'")
-
-    # Start a transaction in the primary. Leave the transaction open.
-    #
-    # The transaction has some subtransactions, but not too many to cause the
-    # CLOG-scanning mechanism to give up.
-    primary_cur.execute("begin")
-    primary_cur.execute("select create_subxacts(50)")
-
-    # Wait for the WAL to be flushed
-    primary_cur.execute("select neon_xlogflush()")
-    wait_for_last_flush_lsn(env, primary, env.initial_tenant, env.initial_timeline)
-
-    if shutdown:
-        primary.stop(mode="fast")
-
-    # Create a replica. It should start up normally, thanks to the CLOG-scanning
-    # mechanism.
-    secondary = env.endpoints.new_replica_start(
-        origin=primary, endpoint_id="secondary", config_lines=["max_prepared_transactions=5"]
-    )
-
-    # The transaction did not commit, so it should not be visible in the secondary
-    secondary_conn = secondary.connect()
-    secondary_cur = secondary_conn.cursor()
-    secondary_cur.execute("select count(*) from t")
-    assert secondary_cur.fetchone() == (0,)
-    secondary_cur.execute("select count(*) from t1")
-    assert secondary_cur.fetchone() == (0,)
-    secondary_cur.execute("select count(*) from t2")
-    assert secondary_cur.fetchone() == (0,)
-
-    if shutdown:
-        primary.start()
-        primary_conn = primary.connect()
-        primary_cur = primary_conn.cursor()
-    else:
-        primary_cur.execute("commit")
-    primary_cur.execute("commit prepared 't1'")
-    primary_cur.execute("commit prepared 't2'")
-
-    wait_replica_caughtup(primary, secondary)
-
-    secondary_cur.execute("select count(*) from t")
-    if shutdown:
-        assert secondary_cur.fetchone() == (0,)
-    else:
-        assert secondary_cur.fetchone() == (50,)
-    secondary_cur.execute("select * from t1")
-    assert secondary_cur.fetchall() == [(1,)]
-    secondary_cur.execute("select * from t2")
-    assert secondary_cur.fetchall() == [(2,), (3,)]
-
-
-def test_replica_start_with_prepared_xacts_with_subxacts(neon_simple_env: NeonEnv):
-    """
-    Test the CLOG-scanning mechanism at hot standby startup in the presence of
-    prepared transactions, with subtransactions.
-    """
-
-    # Initialize the primary, a test table, and a helper function to create lots
-    # of subtransactions.
-    env = neon_simple_env
-    primary = env.endpoints.create_start(
-        branch_name="main", endpoint_id="primary", config_lines=["max_prepared_transactions=5"]
-    )
-    primary_conn = primary.connect()
-    primary_cur = primary_conn.cursor()
-
-    # Install extension containing function needed for test
-    primary_cur.execute("CREATE EXTENSION neon_test_utils")
-
-    primary_cur.execute("create table t(pk serial primary key, payload integer)")
-    primary_cur.execute(CREATE_SUBXACTS_FUNC)
-
-    # Advance nextXid close to the beginning of the next pg_subtrans segment (2^16 XIDs)
-    #
-    # This is interesting, because it tests that pg_subtrans is initialized correctly
-    # at standby startup. (We had a bug where it didn't at one point during development.)
-    while True:
-        xid = int(query_scalar(primary_cur, "SELECT txid_current()"))
-        log.info(f"xid now {xid}")
-        # Consume 500 transactions at a time until we get close
-        if xid < 65535 - 600:
-            primary_cur.execute("select test_consume_xids(500);")
-        else:
-            break
-    primary_cur.execute("checkpoint")
-
-    # Prepare a transaction for two-phase commit
-    primary_cur.execute("begin")
-    primary_cur.execute("select create_subxacts(1000)")
-    primary_cur.execute("prepare transaction 't1'")
-
-    # Wait for the WAL to be flushed, and stop the primary
-    wait_for_last_flush_lsn(env, primary, env.initial_tenant, env.initial_timeline)
-    primary.stop(mode="fast")
-
-    # Create a replica. It should start up normally, thanks to the CLOG-scanning
-    # mechanism.
-    secondary = env.endpoints.new_replica_start(
-        origin=primary, endpoint_id="secondary", config_lines=["max_prepared_transactions=5"]
-    )
-
-    # The transaction did not commit, so it should not be visible in the secondary
-    secondary_conn = secondary.connect()
-    secondary_cur = secondary_conn.cursor()
-    secondary_cur.execute("select count(*) from t")
-    assert secondary_cur.fetchone() == (0,)
-
-    primary.start()
-
-    # Open a lot of subtransactions in the primary, causing the subxids cache to overflow
-    primary_conn = primary.connect()
-    primary_cur = primary_conn.cursor()
-    primary_cur.execute("select create_subxacts(100000)")
-
-    wait_replica_caughtup(primary, secondary)
-
-    secondary_cur.execute("select count(*) from t")
-    assert secondary_cur.fetchone() == (100000,)
-
-    primary_cur.execute("commit prepared 't1'")
-
-    wait_replica_caughtup(primary, secondary)
-    secondary_cur.execute("select count(*) from t")
-    assert secondary_cur.fetchone() == (101000,)
-
-
-def test_replica_start_with_prepared_xacts_with_many_subxacts(neon_simple_env: NeonEnv):
-    """
-    Test the CLOG-scanning mechanism at hot standby startup in the presence of
-    prepared transactions, with lots of subtransactions.
-
-    Like test_replica_start_with_prepared_xacts_with_subxacts, but with more
-    subxacts, to test that the prepared transaction's subxids don't consume
-    space in the known-assigned XIDs array. (They are set in pg_subtrans
-    instead)
-    """
-
-    # Initialize the primary, a test table, and a helper function to create lots
-    # of subtransactions.
-    env = neon_simple_env
-    primary = env.endpoints.create_start(
-        branch_name="main", endpoint_id="primary", config_lines=["max_prepared_transactions=5"]
-    )
-    primary_conn = primary.connect()
-    primary_cur = primary_conn.cursor()
-
-    # Install extension containing function needed for test
-    primary_cur.execute("CREATE EXTENSION neon_test_utils")
-
-    primary_cur.execute("create table t(pk serial primary key, payload integer)")
-    primary_cur.execute(CREATE_SUBXACTS_FUNC)
-
-    # Prepare a transaction for two-phase commit, with lots of subxids
-    primary_cur.execute("begin")
-    primary_cur.execute("select create_subxacts(50000)")
-
-    # to make things a bit more varied, intersperse a few other XIDs in between
-    # the prepared transaction's sub-XIDs
-    with primary.connect().cursor() as primary_cur2:
-        primary_cur2.execute("insert into t (payload) values (123)")
-        primary_cur2.execute("begin; insert into t (payload) values (-1); rollback")
-
-    primary_cur.execute("select create_subxacts(50000)")
-    primary_cur.execute("prepare transaction 't1'")
-
-    # Wait for the WAL to be flushed
-    wait_for_last_flush_lsn(env, primary, env.initial_tenant, env.initial_timeline)
-
-    primary.stop(mode="fast")
-
-    # Create a replica. It should start up normally, thanks to the CLOG-scanning
-    # mechanism.
-    secondary = env.endpoints.new_replica_start(
-        origin=primary, endpoint_id="secondary", config_lines=["max_prepared_transactions=5"]
-    )
-
-    # The transaction did not commit, so it should not be visible in the secondary
-    secondary_conn = secondary.connect()
-    secondary_cur = secondary_conn.cursor()
-    secondary_cur.execute("select count(*) from t")
-    assert secondary_cur.fetchone() == (1,)
-
-    primary.start()
-
-    # Open a lot of subtransactions in the primary, causing the subxids cache to overflow
-    primary_conn = primary.connect()
-    primary_cur = primary_conn.cursor()
-    primary_cur.execute("select create_subxacts(100000)")
-
-    wait_replica_caughtup(primary, secondary)
-
-    secondary_cur.execute("select count(*) from t")
-    assert secondary_cur.fetchone() == (100001,)
-
-    primary_cur.execute("commit prepared 't1'")
-
-    wait_replica_caughtup(primary, secondary)
-    secondary_cur.execute("select count(*) from t")
-    assert secondary_cur.fetchone() == (200001,)

test_runner/regress/test_replication_start.py

@@ -0,0 +1,32 @@
+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
+
+    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 integer primary key, payload integer)")
+                p_cur.execute("insert into t values (generate_series(1,100000), 0)")
+                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)

vendor/revisions.json

@@ -1,5 +1,5 @@
 {
-  "v16": ["16.3", "b810fdfcbb59afea7ea7bbe0cf94eaccb55a2ea2"],
-  "v15": ["15.7", "4874c8e52ed349a9f8290bbdcd91eb92677a5d24"],
-  "v14": ["14.12", "ad73770c446ea361f43e4f0404798b7e5e7a62d8"]
+  "v16": ["16.3", "e06bebc75306b583e758b52c95946d41109239b2"],
+  "v15": ["15.7", "f54d7373eb0de5a54bce2becdb1c801026c7edff"],
+  "v14": ["14.12", "223dd925959f8124711dd3d867dc8ba6629d52c0"]
 }