Merge branch 'problame/repartition-bail-on-concurrent-call' into problame/avoid-count-deltas-if-no-changes

This PR enforces aspects of `Timeline::repartition` that were already
true at runtime:

- it's not called concurrently, so, bail out if it is anyway (see
  comment why it's not called concurrently)
- the `lsn` should never be moving backwards over the lifetime of a
  Timeline object, because last_record_lsn() can only move forwards
  over the lifetime of a Timeline object

part of #6861

pageserver/src/pgdatadir_mapping.rs

@@ -36,8 +36,6 @@ use tracing::{debug, trace, warn};
 use utils::bin_ser::DeserializeError;
 use utils::{bin_ser::BeSer, lsn::Lsn};
 
-const MAX_AUX_FILE_DELTAS: usize = 1024;
-
 #[derive(Debug)]
 pub enum LsnForTimestamp {
     /// Found commits both before and after the given timestamp
@@ -1405,20 +1403,16 @@ impl<'a> DatadirModification<'a> {
 
         let dir = if let Some(mut dir) = self.pending_aux_files.take() {
             // We already updated aux files in `self`: emit a delta and update our latest value
-            dir.upsert(file_path.clone(), content.clone());
-            if dir.files.len() % MAX_AUX_FILE_DELTAS == 0 {
-                self.put(
-                    AUX_FILES_KEY,
-                    Value::Image(Bytes::from(
-                        AuxFilesDirectory::ser(&dir).context("serialize")?,
-                    )),
-                );
-            } else {
-                self.put(
-                    AUX_FILES_KEY,
-                    Value::WalRecord(NeonWalRecord::AuxFile { file_path, content }),
-                );
-            }
+
+            self.put(
+                AUX_FILES_KEY,
+                Value::WalRecord(NeonWalRecord::AuxFile {
+                    file_path: file_path.clone(),
+                    content: content.clone(),
+                }),
+            );
+
+            dir.upsert(file_path, content);
             dir
         } else {
             // Check if the AUX_FILES_KEY is initialized

pageserver/src/tenant/layer_map.rs

@@ -61,6 +61,8 @@ use utils::lsn::Lsn;
 use historic_layer_coverage::BufferedHistoricLayerCoverage;
 pub use historic_layer_coverage::LayerKey;
 
+pub(crate) use self::historic_layer_coverage::RebuildVersion;
+
 use super::storage_layer::PersistentLayerDesc;
 
 ///
@@ -500,7 +502,7 @@ impl LayerMap {
     ///
     /// Helper function for BatchedUpdates::remove_historic
     ///
-    pub fn remove_historic_noflush(&mut self, layer_desc: &PersistentLayerDesc) {
+    pub(self) fn remove_historic_noflush(&mut self, layer_desc: &PersistentLayerDesc) {
         self.historic
             .remove(historic_layer_coverage::LayerKey::from(layer_desc));
         let layer_key = layer_desc.key();
@@ -525,6 +527,10 @@ impl LayerMap {
         self.historic.rebuild();
     }
 
+    pub fn get_rebuild_version(&self) -> RebuildVersion {
+        self.historic.get_rebuild_version()
+    }
+
     /// Is there a newer image layer for given key- and LSN-range? Or a set
     /// of image layers within the specified lsn range that cover the entire
     /// specified key range?

pageserver/src/tenant/layer_map/historic_layer_coverage.rs

@@ -413,6 +413,8 @@ fn test_persistent_overlapping() {
 /// See this for more on persistent and retroactive techniques:
 /// <https://www.youtube.com/watch?v=WqCWghETNDc&t=581s>
 pub struct BufferedHistoricLayerCoverage<Value> {
+    rebuild_version: RebuildVersion,
+
     /// A persistent layer map that we rebuild when we need to retroactively update
     historic_coverage: HistoricLayerCoverage<Value>,
 
@@ -438,9 +440,33 @@ impl<T: Clone> Default for BufferedHistoricLayerCoverage<T> {
     }
 }
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
+pub struct RebuildVersion(u64);
+
+impl RebuildVersion {
+    fn inc(&mut self) {
+        self.0
+            .checked_add(1)
+            .expect("at current clock cycles, we won't hit this");
+    }
+}
+
+impl Default for RebuildVersion {
+    fn default() -> Self {
+        RebuildVersion(1)
+    }
+}
+
+impl std::fmt::Display for RebuildVersion {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+
 impl<Value: Clone> BufferedHistoricLayerCoverage<Value> {
     pub fn new() -> Self {
         Self {
+            rebuild_version: RebuildVersion::default(),
             historic_coverage: HistoricLayerCoverage::<Value>::new(),
             buffer: BTreeMap::new(),
             layers: BTreeMap::new(),
@@ -462,6 +488,8 @@ impl<Value: Clone> BufferedHistoricLayerCoverage<Value> {
             None => return, // No need to rebuild if buffer is empty
         };
 
+        self.rebuild_version.inc();
+
         // Apply buffered updates to self.layers
         let num_updates = self.buffer.len();
         self.buffer.retain(|layer_key, layer| {
@@ -493,11 +521,17 @@ impl<Value: Clone> BufferedHistoricLayerCoverage<Value> {
 
         // TODO maybe only warn if ratio is at least 10
         info!(
+            version = %self.rebuild_version,
             "Rebuilt layer map. Did {} insertions to process a batch of {} updates.",
-            num_inserted, num_updates,
+            num_inserted,
+            num_updates,
         )
     }
 
+    pub fn get_rebuild_version(&self) -> RebuildVersion {
+        self.rebuild_version
+    }
+
     /// Iterate all the layers
     pub fn iter(&self) -> impl '_ + Iterator<Item = Value> {
         // NOTE we can actually perform this without rebuilding,

pageserver/src/tenant/timeline.rs

@@ -111,10 +111,10 @@ use self::layer_manager::LayerManager;
 use self::logical_size::LogicalSize;
 use self::walreceiver::{WalReceiver, WalReceiverConf};
 
-use super::remote_timeline_client::RemoteTimelineClient;
 use super::secondary::heatmap::{HeatMapLayer, HeatMapTimeline};
 use super::{config::TenantConf, storage_layer::ReadableLayerDesc};
 use super::{debug_assert_current_span_has_tenant_and_timeline_id, AttachedTenantConf};
+use super::{layer_map, remote_timeline_client::RemoteTimelineClient};
 use super::{remote_timeline_client::index::IndexPart, storage_layer::LayerFringe};
 
 #[derive(Debug, PartialEq, Eq, Clone, Copy)]
@@ -210,6 +210,10 @@ pub struct Timeline {
     /// so that e.g. on-demand-download/eviction, and layer spreading, can operate just on `LayerFileManager`.
     pub(crate) layers: Arc<tokio::sync::RwLock<LayerManager>>,
 
+    /// State that [`Self::create_image_layers`] keeps across invocations to determine if it can
+    /// skip an invocation becaucse nothing changed.
+    create_image_layers_skipper_state: std::sync::Mutex<Option<CreateImageLayersParams>>,
+
     /// Set of key ranges which should be covered by image layers to
     /// allow GC to remove old layers. This set is created by GC and its cutoff LSN is also stored.
     /// It is used by compaction task when it checks if new image layer should be created.
@@ -302,8 +306,9 @@ pub struct Timeline {
     // though let's keep them both for better error visibility.
     pub initdb_lsn: Lsn,
 
-    /// When did we last calculate the partitioning?
-    partitioning: Mutex<(KeyPartitioning, Lsn)>,
+    /// Used by [`Timline::repartition`] to avoid re-computing partitioning on every iteration.
+    /// Must bump [`CompactionKeyspacePartitioning::version`] when changing.
+    partitioning: tokio::sync::Mutex<CompactionKeyspacePartitioning>,
 
     /// Configuration: how often should the partitioning be recalculated.
     repartition_threshold: u64,
@@ -400,6 +405,57 @@ pub struct GcInfo {
     pub pitr_cutoff: Lsn,
 }
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+struct CreateImageLayersParams {
+    /// From [`layer_map::LayerMap::get_rebuild_version`].
+    layer_map_version: layer_map::RebuildVersion,
+    /// From [`Timeline::partitioning`].
+    partitioning_version: CompactionKeyspacePartitioningVersion,
+}
+
+/// The version of the keyspace partitioning maintained in [`Timeline::partitioning`].
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+struct CompactionKeyspacePartitioningVersion(u64);
+
+impl CompactionKeyspacePartitioningVersion {
+    fn inc(&mut self) {
+        self.0
+            .checked_add(1)
+            .expect("at current clock cycles, we won't hit this");
+    }
+}
+
+#[derive(Clone)]
+struct CompactionKeyspacePartitioning {
+    version: CompactionKeyspacePartitioningVersion,
+    partitioning: KeyPartitioning,
+    lsn: Lsn,
+}
+
+struct CompactionKeyspacePartition<'a> {
+    key_space: &'a KeySpace,
+    lsn: Lsn,
+}
+
+impl CompactionKeyspacePartitioning {
+    pub fn update(&mut self, partitioning: KeyPartitioning, lsn: Lsn) {
+        assert!(self.lsn <= lsn);
+        self.version.inc();
+        self.partitioning = partitioning;
+        self.lsn = lsn;
+    }
+
+    pub fn iter(&self) -> impl Iterator<Item = CompactionKeyspacePartition<'_>> {
+        self.partitioning
+            .parts
+            .iter()
+            .map(|key_space| CompactionKeyspacePartition {
+                key_space,
+                lsn: self.lsn,
+            })
+    }
+}
+
 /// An error happened in a get() operation.
 #[derive(thiserror::Error, Debug)]
 pub(crate) enum PageReconstructError {
@@ -1154,7 +1210,7 @@ impl Timeline {
             )
             .await
         {
-            Ok((partitioning, lsn)) => {
+            Ok(partitioning) => {
                 // Disables access_stats updates, so that the files we read remain candidates for eviction after we're done with them
                 let image_ctx = RequestContextBuilder::extend(ctx)
                     .access_stats_behavior(AccessStatsBehavior::Skip)
@@ -1168,7 +1224,7 @@ impl Timeline {
                 // 3. Create new image layers for partitions that have been modified
                 // "enough".
                 let layers = self
-                    .create_image_layers(&partitioning, lsn, false, &image_ctx)
+                    .create_image_layers(&partitioning, false, &image_ctx)
                     .await
                     .map_err(anyhow::Error::from)?;
                 if let Some(remote_client) = &self.remote_client {
@@ -1659,8 +1715,15 @@ impl Timeline {
                     // initial logical size is 0.
                     LogicalSize::empty_initial()
                 },
-                partitioning: Mutex::new((KeyPartitioning::new(), Lsn(0))),
+                partitioning: tokio::sync::Mutex::new(CompactionKeyspacePartitioning {
+                    // The other fields don't matter because this is 0 and hence repartition(),
+                    // will calculate a new one
+                    lsn: Lsn(0),
+                    version: CompactionKeyspacePartitioningVersion(0),
+                    partitioning: KeyPartitioning::new(),
+                }),
                 repartition_threshold: 0,
+                create_image_layers_skipper_state: std::sync::Mutex::new(None),
 
                 last_received_wal: Mutex::new(None),
                 rel_size_cache: RwLock::new(HashMap::new()),
@@ -3151,7 +3214,7 @@ impl Timeline {
                 }
                 // Note: The 'ctx' in use here has DownloadBehavior::Error. We should not
                 // require downloading anything during initial import.
-                let (partitioning, _lsn) = self
+                let partitioning = self
                     .repartition(
                         self.initdb_lsn,
                         self.get_compaction_target_size(),
@@ -3166,8 +3229,7 @@ impl Timeline {
 
                 // For image layers, we add them immediately into the layer map.
                 (
-                    self.create_image_layers(&partitioning, self.initdb_lsn, true, ctx)
-                        .await?,
+                    self.create_image_layers(&partitioning, true, ctx).await?,
                     None,
                 )
             } else {
@@ -3405,36 +3467,46 @@ impl Timeline {
         partition_size: u64,
         flags: EnumSet<CompactFlags>,
         ctx: &RequestContext,
-    ) -> anyhow::Result<(KeyPartitioning, Lsn)> {
-        {
-            let partitioning_guard = self.partitioning.lock().unwrap();
-            let distance = lsn.0 - partitioning_guard.1 .0;
-            if partitioning_guard.1 != Lsn(0)
-                && distance <= self.repartition_threshold
-                && !flags.contains(CompactFlags::ForceRepartition)
-            {
-                debug!(
-                    distance,
-                    threshold = self.repartition_threshold,
-                    "no repartitioning needed"
-                );
-                return Ok((partitioning_guard.0.clone(), partitioning_guard.1));
-            }
+    ) -> anyhow::Result<CompactionKeyspacePartitioning> {
+        let Ok(mut partitioning_guard) = self.partitioning.try_lock() else {
+            // NB: there are two callers, one is the compaction task, of which there is only one per struct Tenant and hence Timelien.
+            // The other is the initdb optimization in flush_frozen_layer, used by `boostrap_timeline`, which runs before `.activate()`
+            // and hence before the compaction task starts.
+            anyhow::bail!("repartition() called concurrently, this should not happen");
+        };
+        if lsn < partitioning_guard.lsn {
+            anyhow::bail!("repartition() called with LSN going backwards, this should not happen");
         }
+
+        let distance = lsn.0 - partitioning_guard.lsn.0;
+        if partitioning_guard.lsn != Lsn(0)
+            && distance <= self.repartition_threshold
+            && !flags.contains(CompactFlags::ForceRepartition)
+        {
+            debug!(
+                distance,
+                threshold = self.repartition_threshold,
+                "no repartitioning needed"
+            );
+            return Ok(partitioning_guard.clone());
+        }
+
         let keyspace = self.collect_keyspace(lsn, ctx).await?;
         let partitioning = keyspace.partition(partition_size);
 
-        let mut partitioning_guard = self.partitioning.lock().unwrap();
-        if lsn > partitioning_guard.1 {
-            *partitioning_guard = (partitioning, lsn);
-        } else {
-            warn!("Concurrent repartitioning of keyspace. This unexpected, but probably harmless");
-        }
-        Ok((partitioning_guard.0.clone(), partitioning_guard.1))
+        partitioning_guard.update(partitioning, lsn);
+
+        Ok(partitioning_guard.clone())
     }
 
     // Is it time to create a new image layer for the given partition?
-    async fn time_for_new_image_layer(&self, partition: &KeySpace, lsn: Lsn) -> bool {
+    async fn time_for_new_image_layer(
+        &self,
+        &CompactionKeyspacePartition {
+            key_space: partition,
+            lsn,
+        }: &CompactionKeyspacePartition<'_>,
+    ) -> bool {
         let threshold = self.get_image_creation_threshold();
 
         let guard = self.layers.read().await;
@@ -3509,14 +3581,14 @@ impl Timeline {
         false
     }
 
-    #[tracing::instrument(skip_all, fields(%lsn, %force))]
+    #[tracing::instrument(skip_all, fields(lsn=%partitioning.lsn, %force))]
     async fn create_image_layers(
         self: &Arc<Timeline>,
-        partitioning: &KeyPartitioning,
-        lsn: Lsn,
+        partitioning: &CompactionKeyspacePartitioning,
         force: bool,
         ctx: &RequestContext,
     ) -> Result<Vec<ResidentLayer>, CreateImageLayersError> {
+        // REVIEW: should we not even bump this timer if we're taking short exit?
         let timer = self.metrics.create_images_time_histo.start_timer();
         let mut image_layers = Vec::new();
 
@@ -3531,9 +3603,30 @@ impl Timeline {
         // image layers  <100000000..100000099> and <200000000..200000199> are not completely covering it.
         let mut start = Key::MIN;
 
-        for partition in partitioning.parts.iter() {
-            let img_range = start..partition.ranges.last().unwrap().end;
-            if !force && !self.time_for_new_image_layer(partition, lsn).await {
+        let new_inputs = {
+            let layer_manager = self.layers.read().await;
+            let layer_map = layer_manager.layer_map();
+            CreateImageLayersParams {
+                layer_map_version: layer_map.get_rebuild_version(),
+                partitioning_version: partitioning.version,
+            }
+        };
+        let last_inputs = {
+            // bail out early if nothing changed
+            let last = self.create_image_layers_skipper_state.lock().unwrap();
+            match (force, &*last, &new_inputs) {
+                (false, Some(last), new_inputs) if last == new_inputs => {
+                    debug!(?last, ?new_inputs, "inputs to image layer creation algorithm have not changed since last invocation");
+                    return Ok(image_layers);
+                }
+                _ => (), // we'll update .last once we finish with success
+            }
+            *last
+        };
+
+        for partition in partitioning.iter() {
+            let img_range = start..partition.key_space.ranges.last().unwrap().end;
+            if !force && !self.time_for_new_image_layer(&partition).await {
                 start = img_range.end;
                 continue;
             }
@@ -3543,7 +3636,7 @@ impl Timeline {
                 self.timeline_id,
                 self.tenant_shard_id,
                 &img_range,
-                lsn,
+                partitioning.lsn,
             )
             .await?;
 
@@ -3556,7 +3649,7 @@ impl Timeline {
             let mut wrote_keys = false;
 
             let mut key_request_accum = KeySpaceAccum::new();
-            for range in &partition.ranges {
+            for range in &partition.key_space.ranges {
                 let mut key = range.start;
                 while key < range.end {
                     // Decide whether to retain this key: usually we do, but sharded tenants may
@@ -3580,7 +3673,11 @@ impl Timeline {
                         || last_key_in_range
                     {
                         let results = self
-                            .get_vectored(key_request_accum.consume_keyspace(), lsn, ctx)
+                            .get_vectored(
+                                key_request_accum.consume_keyspace(),
+                                partitioning.lsn,
+                                ctx,
+                            )
                             .await?;
 
                         for (img_key, img) in results {
@@ -3669,6 +3766,25 @@ impl Timeline {
             .context("fsync of timeline dir")?;
         }
 
+        // Remember the inputs so that we take the early exit next compaction iteration
+        // if nothing changed in the meantime.
+        // NB: since we created `new_inputs`, the layer map might have changed and hence
+        // `get_rebuild_version()` might have advanced already, e.g., by a
+        // concurrent garbage collection iteration that removed layers. That's ok, next
+        // `new_inputs` will observe an increased `get_rebuild_version()` and run again.
+        // Same goes for the modifications that we're doing below: if `image_layers` is
+        // not empty, we'll insert them into the layer map in the code below, which
+        // will bump `get_rebuild_version()`, which will make us re-run once more.
+        //
+        {
+            let mut state = self.create_image_layers_skipper_state.lock().unwrap();
+            if *state != last_inputs {
+                warn!(?last_inputs, ?new_inputs, observed=?*state, "unexpected: create_image_layers called concurrently? not caching inputs");
+            } else {
+                *state = Some(new_inputs);
+            }
+        }
+
         let mut guard = self.layers.write().await;
 
         // FIXME: we could add the images to be uploaded *before* returning from here, but right

pgxn/neon/control_plane_connector.c

@@ -35,16 +35,16 @@
 #include "utils/memutils.h"
 #include "utils/jsonb.h"
 
-#include "neon_utils.h"
-
 static ProcessUtility_hook_type PreviousProcessUtilityHook = NULL;
 
-static const char *jwt_token = NULL;
-
 /* GUCs */
 static char *ConsoleURL = NULL;
 static bool ForwardDDL = true;
 
+/* Curl structures for sending the HTTP requests */
+static CURL *CurlHandle;
+static struct curl_slist *ContentHeader = NULL;
+
 /*
  * CURL docs say that this buffer must exist until we call curl_easy_cleanup
  * (which we never do), so we make this a static
@@ -226,8 +226,6 @@ ErrorWriteCallback(char *ptr, size_t size, size_t nmemb, void *userdata)
 static void
 SendDeltasToControlPlane()
 {
-	static CURL		*handle = NULL;
-
 	if (!RootTable.db_table && !RootTable.role_table)
 		return;
 	if (!ConsoleURL)
@@ -238,57 +236,29 @@ SendDeltasToControlPlane()
 	if (!ForwardDDL)
 		return;
 
-	if (handle == NULL)
-	{
-		struct curl_slist *headers = NULL;
-
-		headers = curl_slist_append(headers, "Content-Type: application/json");
-		if (headers == NULL)
-		{
-			elog(ERROR, "Failed to set Content-Type header");
-		}
-
-		if (jwt_token)
-		{
-			char		auth_header[8192];
-
-			snprintf(auth_header, sizeof(auth_header), "Authorization: Bearer %s", jwt_token);
-			headers = curl_slist_append(headers, auth_header);
-			if (headers == NULL)
-			{
-				elog(ERROR, "Failed to set Authorization header");
-			}
-		}
-
-		handle = alloc_curl_handle();
-
-		curl_easy_setopt(handle, CURLOPT_CUSTOMREQUEST, "PATCH");
-		curl_easy_setopt(handle, CURLOPT_HTTPHEADER, headers);
-		curl_easy_setopt(handle, CURLOPT_URL, ConsoleURL);
-		curl_easy_setopt(handle, CURLOPT_ERRORBUFFER, CurlErrorBuf);
-		curl_easy_setopt(handle, CURLOPT_TIMEOUT, 3L /* seconds */ );
-		curl_easy_setopt(handle, CURLOPT_WRITEFUNCTION, ErrorWriteCallback);
-	}
-
 	char	   *message = ConstructDeltaMessage();
-	ErrorString str;
+	ErrorString str = {};
 
-	str.size = 0;
-
-	curl_easy_setopt(handle, CURLOPT_POSTFIELDS, message);
-	curl_easy_setopt(handle, CURLOPT_WRITEDATA, &str);
+	curl_easy_setopt(CurlHandle, CURLOPT_CUSTOMREQUEST, "PATCH");
+	curl_easy_setopt(CurlHandle, CURLOPT_HTTPHEADER, ContentHeader);
+	curl_easy_setopt(CurlHandle, CURLOPT_POSTFIELDS, message);
+	curl_easy_setopt(CurlHandle, CURLOPT_URL, ConsoleURL);
+	curl_easy_setopt(CurlHandle, CURLOPT_ERRORBUFFER, CurlErrorBuf);
+	curl_easy_setopt(CurlHandle, CURLOPT_TIMEOUT, 3L /* seconds */ );
+	curl_easy_setopt(CurlHandle, CURLOPT_WRITEDATA, &str);
+	curl_easy_setopt(CurlHandle, CURLOPT_WRITEFUNCTION, ErrorWriteCallback);
 
 	const int	num_retries = 5;
-	CURLcode	curl_status;
+	int			curl_status;
 
 	for (int i = 0; i < num_retries; i++)
 	{
-		if ((curl_status = curl_easy_perform(handle)) == 0)
+		if ((curl_status = curl_easy_perform(CurlHandle)) == 0)
 			break;
 		elog(LOG, "Curl request failed on attempt %d: %s", i, CurlErrorBuf);
 		pg_usleep(1000 * 1000);
 	}
-	if (curl_status != CURLE_OK)
+	if (curl_status != 0)
 	{
 		elog(ERROR, "Failed to perform curl request: %s", CurlErrorBuf);
 	}
@@ -296,11 +266,13 @@ SendDeltasToControlPlane()
 	{
 		long		response_code;
 
-		if (curl_easy_getinfo(handle, CURLINFO_RESPONSE_CODE, &response_code) != CURLE_UNKNOWN_OPTION)
+		if (curl_easy_getinfo(CurlHandle, CURLINFO_RESPONSE_CODE, &response_code) != CURLE_UNKNOWN_OPTION)
 		{
+			bool		error_exists = str.size != 0;
+
 			if (response_code != 200)
 			{
-				if (str.size != 0)
+				if (error_exists)
 				{
 					elog(ERROR,
 						 "Received HTTP code %ld from control plane: %s",
@@ -863,10 +835,34 @@ InitControlPlaneConnector()
 							 NULL,
 							 NULL);
 
-	jwt_token = getenv("NEON_CONTROL_PLANE_TOKEN");
+	const char *jwt_token = getenv("NEON_CONTROL_PLANE_TOKEN");
+
 	if (!jwt_token)
 	{
 		elog(LOG, "Missing NEON_CONTROL_PLANE_TOKEN environment variable, forwarding will not be authenticated");
 	}
 
+	if (curl_global_init(CURL_GLOBAL_DEFAULT))
+	{
+		elog(ERROR, "Failed to initialize curl");
+	}
+	if ((CurlHandle = curl_easy_init()) == NULL)
+	{
+		elog(ERROR, "Failed to initialize curl handle");
+	}
+	if ((ContentHeader = curl_slist_append(ContentHeader, "Content-Type: application/json")) == NULL)
+	{
+		elog(ERROR, "Failed to initialize content header");
+	}
+
+	if (jwt_token)
+	{
+		char		auth_header[8192];
+
+		snprintf(auth_header, sizeof(auth_header), "Authorization: Bearer %s", jwt_token);
+		if ((ContentHeader = curl_slist_append(ContentHeader, auth_header)) == NULL)
+		{
+			elog(ERROR, "Failed to initialize authorization header");
+		}
+	}
 }

pgxn/neon/extension_server.c

@@ -14,8 +14,6 @@
 
 #include "utils/guc.h"
 
-#include "neon_utils.h"
-
 static int	extension_server_port = 0;
 
 static download_extension_file_hook_type prev_download_extension_file_hook = NULL;
@@ -33,19 +31,15 @@ static download_extension_file_hook_type prev_download_extension_file_hook = NUL
 static bool
 neon_download_extension_file_http(const char *filename, bool is_library)
 {
-	static CURL	   *handle = NULL;
-
+	CURL	   *curl;
 	CURLcode	res;
 	char	   *compute_ctl_url;
 	char	   *postdata;
 	bool		ret = false;
 
-	if (handle == NULL)
+	if ((curl = curl_easy_init()) == NULL)
 	{
-		handle = alloc_curl_handle();
-
-		curl_easy_setopt(handle, CURLOPT_CUSTOMREQUEST, "POST");
-		curl_easy_setopt(handle, CURLOPT_TIMEOUT, 3L /* seconds */ );
+		elog(ERROR, "Failed to initialize curl handle");
 	}
 
 	compute_ctl_url = psprintf("http://localhost:%d/extension_server/%s%s",
@@ -53,22 +47,28 @@ neon_download_extension_file_http(const char *filename, bool is_library)
 
 	elog(LOG, "Sending request to compute_ctl: %s", compute_ctl_url);
 
-	curl_easy_setopt(handle, CURLOPT_URL, compute_ctl_url);
+	curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, "POST");
+	curl_easy_setopt(curl, CURLOPT_URL, compute_ctl_url);
+	curl_easy_setopt(curl, CURLOPT_TIMEOUT, 3L /* seconds */ );
 
-	/* Perform the request, res will get the return code */
-	res = curl_easy_perform(handle);
-	/* Check for errors */
-	if (res == CURLE_OK)
+	if (curl)
 	{
-		ret = true;
-	}
-	else
-	{
-		/*
-		 * Don't error here because postgres will try to find the file and will
-		 * fail with some proper error message if it's not found.
-		 */
-		elog(WARNING, "neon_download_extension_file_http failed: %s\n", curl_easy_strerror(res));
+		/* Perform the request, res will get the return code */
+		res = curl_easy_perform(curl);
+		/* Check for errors */
+		if (res == CURLE_OK)
+		{
+			ret = true;
+		}
+		else
+		{
+			/* Don't error here because postgres will try to find the file */
+			/* and will fail with some proper error message if it's not found. */
+			elog(WARNING, "neon_download_extension_file_http failed: %s\n", curl_easy_strerror(res));
+		}
+
+		/* always cleanup */
+		curl_easy_cleanup(curl);
 	}
 
 	return ret;

pgxn/neon/neon_utils.c

@@ -1,8 +1,5 @@
-#include <sys/resource.h>
 
-#ifndef WALPROPOSER_LIB
-#include <curl/curl.h>
-#endif
+#include <sys/resource.h>
 
 #include "postgres.h"
 
@@ -117,48 +114,3 @@ disable_core_dump()
 		fprintf(stderr, "WARNING: disable cores setrlimit failed: %s", strerror(save_errno));
 	}
 }
-
-#ifndef WALPROPOSER_LIB
-
-/*
- * On macOS with a libcurl that has IPv6 support, curl_global_init() calls
- * SCDynamicStoreCopyProxies(), which makes the program multithreaded. An ideal
- * place to call curl_global_init() would be _PG_init(), but Neon has to be
- * added to shared_preload_libraries, which are loaded in the Postmaster
- * process. The Postmaster is not supposed to become multithreaded at any point
- * in its lifecycle. Postgres doesn't have any good hook that I know of to
- * initialize per-backend structures, so we have to check this on any
- * allocation of a CURL handle.
- *
- * Free the allocated CURL handle with curl_easy_cleanup(3).
- *
- * https://developer.apple.com/documentation/systemconfiguration/1517088-scdynamicstorecopyproxies
- */
-CURL *
-alloc_curl_handle(void)
-{
-	static bool curl_initialized = false;
-
-	CURL *handle;
-
-	if (unlikely(!curl_initialized))
-	{
-		/* Protected by mutex internally */
-		if (curl_global_init(CURL_GLOBAL_DEFAULT))
-		{
-			elog(ERROR, "Failed to initialize curl");
-		}
-
-		curl_initialized = true;
-	}
-
-	handle = curl_easy_init();
-	if (handle == NULL)
-	{
-		elog(ERROR, "Failed to initialize curl handle");
-	}
-
-	return handle;
-}
-
-#endif

pgxn/neon/neon_utils.h

@@ -1,12 +1,6 @@
 #ifndef __NEON_UTILS_H__
 #define __NEON_UTILS_H__
 
-#include "lib/stringinfo.h"
-
-#ifndef WALPROPOSER_LIB
-#include <curl/curl.h>
-#endif
-
 bool		HexDecodeString(uint8 *result, char *input, int nbytes);
 uint32		pq_getmsgint32_le(StringInfo msg);
 uint64		pq_getmsgint64_le(StringInfo msg);
@@ -14,10 +8,4 @@ void		pq_sendint32_le(StringInfo buf, uint32 i);
 void		pq_sendint64_le(StringInfo buf, uint64 i);
 extern void disable_core_dump();
 
-#ifndef WALPROPOSER_LIB
-
-CURL *		alloc_curl_handle(void);
-
-#endif
-
 #endif							/* __NEON_UTILS_H__ */

safekeeper/src/bin/safekeeper.rs

@@ -166,10 +166,6 @@ struct Args {
     /// useful for debugging.
     #[arg(long)]
     current_thread_runtime: bool,
-    /// Keep horizon for walsenders, i.e. don't remove WAL segments that are
-    /// still needed for existing replication connection.
-    #[arg(long)]
-    walsenders_keep_horizon: bool,
 }
 
 // Like PathBufValueParser, but allows empty string.
@@ -299,7 +295,6 @@ async fn main() -> anyhow::Result<()> {
         pg_tenant_only_auth,
         http_auth,
         current_thread_runtime: args.current_thread_runtime,
-        walsenders_keep_horizon: args.walsenders_keep_horizon,
     };
 
     // initialize sentry if SENTRY_DSN is provided

safekeeper/src/lib.rs

@@ -78,7 +78,6 @@ pub struct SafeKeeperConf {
     pub pg_tenant_only_auth: Option<Arc<JwtAuth>>,
     pub http_auth: Option<Arc<SwappableJwtAuth>>,
     pub current_thread_runtime: bool,
-    pub walsenders_keep_horizon: bool,
 }
 
 impl SafeKeeperConf {
@@ -122,7 +121,6 @@ impl SafeKeeperConf {
             heartbeat_timeout: Duration::new(5, 0),
             max_offloader_lag_bytes: defaults::DEFAULT_MAX_OFFLOADER_LAG_BYTES,
             current_thread_runtime: false,
-            walsenders_keep_horizon: false,
         }
     }
 }

safekeeper/src/safekeeper.rs

@@ -4,7 +4,7 @@ use anyhow::{bail, Context, Result};
 use byteorder::{LittleEndian, ReadBytesExt};
 use bytes::{Buf, BufMut, Bytes, BytesMut};
 
-use postgres_ffi::{TimeLineID, MAX_SEND_SIZE};
+use postgres_ffi::{TimeLineID, XLogSegNo, MAX_SEND_SIZE};
 use serde::{Deserialize, Serialize};
 use std::cmp::max;
 use std::cmp::min;
@@ -946,12 +946,28 @@ where
         }
         Ok(())
     }
+
+    /// Get oldest segno we still need to keep. We hold WAL till it is consumed
+    /// by all of 1) pageserver (remote_consistent_lsn) 2) peers 3) s3
+    /// offloading.
+    /// While it is safe to use inmem values for determining horizon,
+    /// we use persistent to make possible normal states less surprising.
+    pub fn get_horizon_segno(&self, wal_backup_enabled: bool) -> XLogSegNo {
+        let mut horizon_lsn = min(
+            self.state.remote_consistent_lsn,
+            self.state.peer_horizon_lsn,
+        );
+        if wal_backup_enabled {
+            horizon_lsn = min(horizon_lsn, self.state.backup_lsn);
+        }
+        horizon_lsn.segment_number(self.state.server.wal_seg_size as usize)
+    }
 }
 
 #[cfg(test)]
 mod tests {
     use futures::future::BoxFuture;
-    use postgres_ffi::{XLogSegNo, WAL_SEGMENT_SIZE};
+    use postgres_ffi::WAL_SEGMENT_SIZE;
 
     use super::*;
     use crate::{

safekeeper/src/send_wal.rs

@@ -136,21 +136,6 @@ impl WalSenders {
         self.mutex.lock().slots.iter().flatten().cloned().collect()
     }
 
-    /// Get LSN of the most lagging pageserver receiver. Return None if there are no
-    /// active walsenders.
-    pub fn laggard_lsn(self: &Arc<WalSenders>) -> Option<Lsn> {
-        self.mutex
-            .lock()
-            .slots
-            .iter()
-            .flatten()
-            .filter_map(|s| match s.feedback {
-                ReplicationFeedback::Pageserver(feedback) => Some(feedback.last_received_lsn),
-                ReplicationFeedback::Standby(_) => None,
-            })
-            .min()
-    }
-
     /// Get aggregated pageserver feedback.
     pub fn get_ps_feedback(self: &Arc<WalSenders>) -> PageserverFeedback {
         self.mutex.lock().agg_ps_feedback

safekeeper/src/timeline.rs

@@ -286,29 +286,6 @@ impl SharedState {
             .cloned()
             .collect()
     }
-
-    /// Get oldest segno we still need to keep. We hold WAL till it is consumed
-    /// by all of 1) pageserver (remote_consistent_lsn) 2) peers 3) s3
-    /// offloading.
-    /// While it is safe to use inmem values for determining horizon,
-    /// we use persistent to make possible normal states less surprising.
-    fn get_horizon_segno(
-        &self,
-        wal_backup_enabled: bool,
-        extra_horizon_lsn: Option<Lsn>,
-    ) -> XLogSegNo {
-        let state = &self.sk.state;
-
-        use std::cmp::min;
-        let mut horizon_lsn = min(state.remote_consistent_lsn, state.peer_horizon_lsn);
-        if wal_backup_enabled {
-            horizon_lsn = min(horizon_lsn, state.backup_lsn);
-        }
-        if let Some(extra_horizon_lsn) = extra_horizon_lsn {
-            horizon_lsn = min(horizon_lsn, extra_horizon_lsn);
-        }
-        horizon_lsn.segment_number(state.server.wal_seg_size as usize)
-    }
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -376,12 +353,6 @@ pub struct Timeline {
 
     /// Directory where timeline state is stored.
     pub timeline_dir: Utf8PathBuf,
-
-    /// Should we keep WAL on disk for active replication connections.
-    /// Especially useful for sharding, when different shards process WAL
-    /// with different speed.
-    // TODO: add `Arc<SafeKeeperConf>` here instead of adding each field separately.
-    walsenders_keep_horizon: bool,
 }
 
 impl Timeline {
@@ -415,7 +386,6 @@ impl Timeline {
             cancellation_rx,
             cancellation_tx,
             timeline_dir: conf.timeline_dir(&ttid),
-            walsenders_keep_horizon: conf.walsenders_keep_horizon,
         })
     }
 
@@ -448,7 +418,6 @@ impl Timeline {
             cancellation_rx,
             cancellation_tx,
             timeline_dir: conf.timeline_dir(&ttid),
-            walsenders_keep_horizon: conf.walsenders_keep_horizon,
         })
     }
 
@@ -848,20 +817,10 @@ impl Timeline {
             bail!(TimelineError::Cancelled(self.ttid));
         }
 
-        // If enabled, we use LSN of the most lagging walsender as a WAL removal horizon.
-        // This allows to get better read speed for pageservers that are lagging behind,
-        // at the cost of keeping more WAL on disk.
-        let replication_horizon_lsn = if self.walsenders_keep_horizon {
-            self.walsenders.laggard_lsn()
-        } else {
-            None
-        };
-
         let horizon_segno: XLogSegNo;
         let remover = {
             let shared_state = self.write_shared_state().await;
-            horizon_segno =
-                shared_state.get_horizon_segno(wal_backup_enabled, replication_horizon_lsn);
+            horizon_segno = shared_state.sk.get_horizon_segno(wal_backup_enabled);
             if horizon_segno <= 1 || horizon_segno <= shared_state.last_removed_segno {
                 return Ok(()); // nothing to do
             }

safekeeper/tests/walproposer_sim/safekeeper.rs

@@ -175,7 +175,6 @@ pub fn run_server(os: NodeOs, disk: Arc<SafekeeperDisk>) -> Result<()> {
         pg_tenant_only_auth: None,
         http_auth: None,
         current_thread_runtime: false,
-        walsenders_keep_horizon: false,
     };
 
     let mut global = GlobalMap::new(disk, conf.clone())?;

vendor/revisions.json

@@ -1,5 +1,5 @@
 {
-    "postgres-v16": "dc40299045a377ec3b302c900134468a1b0f58ee",
-    "postgres-v15": "0baccce15a3b0446af5c403d2e869a04541b63c4",
-    "postgres-v14": "17101190de8a54b95e0831c66c3da426ed33db34"
+    "postgres-v16": "9c37a4988463a97d9cacb321acf3828b09823269",
+    "postgres-v15": "ca2def999368d9df098a637234ad5a9003189463",
+    "postgres-v14": "9dd9956c55ffbbd9abe77d10382453757fedfcf5"
 }