Perform check for replicaiton LAG only for backends writting WAL

.github/workflows/build_and_test.yml

@@ -780,7 +780,7 @@ jobs:
       run:
         shell: sh -eu {0}
     env:
-      VM_BUILDER_VERSION: v0.16.3
+      VM_BUILDER_VERSION: v0.16.2
 
     steps:
       - name: Checkout
@@ -804,6 +804,7 @@ jobs:
           ./vm-builder \
             -enable-file-cache \
             -enable-monitor \
+            -enable-informant \
             -src=369495373322.dkr.ecr.eu-central-1.amazonaws.com/compute-node-${{ matrix.version }}:${{needs.tag.outputs.build-tag}} \
             -dst=369495373322.dkr.ecr.eu-central-1.amazonaws.com/vm-compute-node-${{ matrix.version }}:${{needs.tag.outputs.build-tag}}
 
@@ -945,7 +946,7 @@ jobs:
         version: [ v14, v15 ]
 
     env:
-      EXTENSIONS_IMAGE: ${{ github.ref_name == 'release' && '093970136003' || '369495373322'}}.dkr.ecr.eu-central-1.amazonaws.com/extensions-${{ matrix.version }}:latest
+      EXTENSIONS_IMAGE: ${{ github.ref_name == 'release' && '093970136003' || '369495373322'}}.dkr.ecr.eu-central-1.amazonaws.com/extensions-${{ matrix.version }}:${{ needs.tag.outputs.build-tag }}
       AWS_ACCESS_KEY_ID: ${{ github.ref_name == 'release' && secrets.AWS_ACCESS_KEY_PROD || secrets.AWS_ACCESS_KEY_DEV }}
       AWS_SECRET_ACCESS_KEY: ${{ github.ref_name == 'release' && secrets.AWS_SECRET_KEY_PROD || secrets.AWS_SECRET_KEY_DEV }}
       S3_BUCKETS: ${{ github.ref_name == 'release' && vars.S3_EXTENSIONS_BUCKETS_PROD || vars.S3_EXTENSIONS_BUCKETS_DEV }}

Cargo.lock

@@ -3569,9 +3569,9 @@ dependencies = [
 
 [[package]]
 name = "rustls-webpki"
-version = "0.100.2"
+version = "0.100.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e98ff011474fa39949b7e5c0428f9b4937eda7da7848bbb947786b7be0b27dab"
+checksum = "d6207cd5ed3d8dca7816f8f3725513a34609c0c765bf652b8c3cb4cfd87db46b"
 dependencies = [
  "ring",
  "untrusted",

deny.toml

@@ -18,7 +18,7 @@ vulnerability = "deny"
 unmaintained = "warn"
 yanked = "warn"
 notice = "warn"
-ignore = ["RUSTSEC-2023-0052"]
+ignore = []
 
 # This section is considered when running `cargo deny check licenses`
 # More documentation for the licenses section can be found here:

pageserver/src/bin/pageserver.rs

@@ -11,7 +11,6 @@ use metrics::launch_timestamp::{set_launch_timestamp_metric, LaunchTimestamp};
 use pageserver::disk_usage_eviction_task::{self, launch_disk_usage_global_eviction_task};
 use pageserver::metrics::{STARTUP_DURATION, STARTUP_IS_LOADING};
 use pageserver::task_mgr::WALRECEIVER_RUNTIME;
-use pageserver::tenant::TenantSharedResources;
 use remote_storage::GenericRemoteStorage;
 use tokio::time::Instant;
 use tracing::*;
@@ -383,10 +382,8 @@ fn start_pageserver(
 
     BACKGROUND_RUNTIME.block_on(mgr::init_tenant_mgr(
         conf,
-        TenantSharedResources {
-            broker_client: broker_client.clone(),
-            remote_storage: remote_storage.clone(),
-        },
+        broker_client.clone(),
+        remote_storage.clone(),
         order,
     ))?;
 

pageserver/src/tenant.rs

@@ -56,7 +56,6 @@ use self::remote_timeline_client::RemoteTimelineClient;
 use self::timeline::uninit::TimelineUninitMark;
 use self::timeline::uninit::UninitializedTimeline;
 use self::timeline::EvictionTaskTenantState;
-use self::timeline::TimelineResources;
 use crate::config::PageServerConf;
 use crate::context::{DownloadBehavior, RequestContext};
 use crate::import_datadir;
@@ -151,14 +150,6 @@ pub const TENANT_ATTACHING_MARKER_FILENAME: &str = "attaching";
 
 pub const TENANT_DELETED_MARKER_FILE_NAME: &str = "deleted";
 
-/// References to shared objects that are passed into each tenant, such
-/// as the shared remote storage client and process initialization state.
-#[derive(Clone)]
-pub struct TenantSharedResources {
-    pub broker_client: storage_broker::BrokerClientChannel,
-    pub remote_storage: Option<GenericRemoteStorage>,
-}
-
 ///
 /// Tenant consists of multiple timelines. Keep them in a hash table.
 ///
@@ -398,7 +389,7 @@ impl Tenant {
     async fn timeline_init_and_sync(
         &self,
         timeline_id: TimelineId,
-        resources: TimelineResources,
+        remote_client: Option<RemoteTimelineClient>,
         remote_startup_data: Option<RemoteStartupData>,
         local_metadata: Option<TimelineMetadata>,
         ancestor: Option<Arc<Timeline>>,
@@ -419,7 +410,7 @@ impl Tenant {
             timeline_id,
             up_to_date_metadata,
             ancestor.clone(),
-            resources,
+            remote_client,
             init_order,
             CreateTimelineCause::Load,
         )?;
@@ -710,22 +701,14 @@ impl Tenant {
                 .expect("just put it in above");
 
             // TODO again handle early failure
-            self.load_remote_timeline(
-                timeline_id,
-                index_part,
-                remote_metadata,
-                TimelineResources {
-                    remote_client: Some(remote_client),
-                },
-                ctx,
-            )
-            .await
-            .with_context(|| {
-                format!(
-                    "failed to load remote timeline {} for tenant {}",
-                    timeline_id, self.tenant_id
-                )
-            })?;
+            self.load_remote_timeline(timeline_id, index_part, remote_metadata, remote_client, ctx)
+                .await
+                .with_context(|| {
+                    format!(
+                        "failed to load remote timeline {} for tenant {}",
+                        timeline_id, self.tenant_id
+                    )
+                })?;
         }
 
         // Walk through deleted timelines, resume deletion
@@ -780,7 +763,7 @@ impl Tenant {
         timeline_id: TimelineId,
         index_part: IndexPart,
         remote_metadata: TimelineMetadata,
-        resources: TimelineResources,
+        remote_client: RemoteTimelineClient,
         ctx: &RequestContext,
     ) -> anyhow::Result<()> {
         span::debug_assert_current_span_has_tenant_id();
@@ -810,7 +793,7 @@ impl Tenant {
 
         self.timeline_init_and_sync(
             timeline_id,
-            resources,
+            Some(remote_client),
             Some(RemoteStartupData {
                 index_part,
                 remote_metadata,
@@ -857,7 +840,8 @@ impl Tenant {
     pub(crate) fn spawn_load(
         conf: &'static PageServerConf,
         tenant_id: TenantId,
-        resources: TenantSharedResources,
+        broker_client: storage_broker::BrokerClientChannel,
+        remote_storage: Option<GenericRemoteStorage>,
         init_order: Option<InitializationOrder>,
         tenants: &'static tokio::sync::RwLock<TenantsMap>,
         ctx: &RequestContext,
@@ -872,9 +856,6 @@ impl Tenant {
             }
         };
 
-        let broker_client = resources.broker_client;
-        let remote_storage = resources.remote_storage;
-
         let wal_redo_manager = Arc::new(PostgresRedoManager::new(conf, tenant_id));
         let tenant = Tenant::new(
             TenantState::Loading,
@@ -1260,9 +1241,16 @@ impl Tenant {
     ) -> Result<(), LoadLocalTimelineError> {
         span::debug_assert_current_span_has_tenant_id();
 
-        let mut resources = self.build_timeline_resources(timeline_id);
+        let remote_client = self.remote_storage.as_ref().map(|remote_storage| {
+            RemoteTimelineClient::new(
+                remote_storage.clone(),
+                self.conf,
+                self.tenant_id,
+                timeline_id,
+            )
+        });
 
-        let (remote_startup_data, remote_client) = match resources.remote_client {
+        let (remote_startup_data, remote_client) = match remote_client {
             Some(remote_client) => match remote_client.download_index_file().await {
                 Ok(index_part) => {
                     let index_part = match index_part {
@@ -1350,10 +1338,9 @@ impl Tenant {
                     return Ok(());
                 }
 
-                (None, resources.remote_client)
+                (None, remote_client)
             }
         };
-        resources.remote_client = remote_client;
 
         let ancestor = if let Some(ancestor_timeline_id) = local_metadata.ancestor_timeline() {
             let ancestor_timeline = self.get_timeline(ancestor_timeline_id, false)
@@ -1366,7 +1353,7 @@ impl Tenant {
 
         self.timeline_init_and_sync(
             timeline_id,
-            resources,
+            remote_client,
             remote_startup_data,
             Some(local_metadata),
             ancestor,
@@ -2238,7 +2225,7 @@ impl Tenant {
         new_timeline_id: TimelineId,
         new_metadata: &TimelineMetadata,
         ancestor: Option<Arc<Timeline>>,
-        resources: TimelineResources,
+        remote_client: Option<RemoteTimelineClient>,
         init_order: Option<&InitializationOrder>,
         cause: CreateTimelineCause,
     ) -> anyhow::Result<Arc<Timeline>> {
@@ -2267,7 +2254,7 @@ impl Tenant {
             new_timeline_id,
             self.tenant_id,
             Arc::clone(&self.walredo_mgr),
-            resources,
+            remote_client,
             pg_version,
             initial_logical_size_can_start.cloned(),
             initial_logical_size_attempt.cloned().flatten(),
@@ -2915,23 +2902,6 @@ impl Tenant {
         Ok(timeline)
     }
 
-    /// Call this before constructing a timeline, to build its required structures
-    fn build_timeline_resources(&self, timeline_id: TimelineId) -> TimelineResources {
-        let remote_client = if let Some(remote_storage) = self.remote_storage.as_ref() {
-            let remote_client = RemoteTimelineClient::new(
-                remote_storage.clone(),
-                self.conf,
-                self.tenant_id,
-                timeline_id,
-            );
-            Some(remote_client)
-        } else {
-            None
-        };
-
-        TimelineResources { remote_client }
-    }
-
     /// Creates intermediate timeline structure and its files.
     ///
     /// An empty layer map is initialized, and new data and WAL can be imported starting
@@ -2948,17 +2918,25 @@ impl Tenant {
     ) -> anyhow::Result<UninitializedTimeline> {
         let tenant_id = self.tenant_id;
 
-        let resources = self.build_timeline_resources(new_timeline_id);
-        if let Some(remote_client) = &resources.remote_client {
+        let remote_client = if let Some(remote_storage) = self.remote_storage.as_ref() {
+            let remote_client = RemoteTimelineClient::new(
+                remote_storage.clone(),
+                self.conf,
+                tenant_id,
+                new_timeline_id,
+            );
             remote_client.init_upload_queue_for_empty_remote(new_metadata)?;
-        }
+            Some(remote_client)
+        } else {
+            None
+        };
 
         let timeline_struct = self
             .create_timeline_struct(
                 new_timeline_id,
                 new_metadata,
                 ancestor,
-                resources,
+                remote_client,
                 None,
                 CreateTimelineCause::Load,
             )

pageserver/src/tenant/mgr.rs

@@ -29,7 +29,6 @@ use utils::id::{TenantId, TimelineId};
 
 use super::delete::DeleteTenantError;
 use super::timeline::delete::DeleteTimelineFlow;
-use super::TenantSharedResources;
 
 /// The tenants known to the pageserver.
 /// The enum variants are used to distinguish the different states that the pageserver can be in.
@@ -67,7 +66,8 @@ static TENANTS: Lazy<RwLock<TenantsMap>> = Lazy::new(|| RwLock::new(TenantsMap::
 #[instrument(skip_all)]
 pub async fn init_tenant_mgr(
     conf: &'static PageServerConf,
-    resources: TenantSharedResources,
+    broker_client: storage_broker::BrokerClientChannel,
+    remote_storage: Option<GenericRemoteStorage>,
     init_order: InitializationOrder,
 ) -> anyhow::Result<()> {
     // Scan local filesystem for attached tenants
@@ -125,7 +125,8 @@ pub async fn init_tenant_mgr(
                     match schedule_local_tenant_processing(
                         conf,
                         &tenant_dir_path,
-                        resources.clone(),
+                        broker_client.clone(),
+                        remote_storage.clone(),
                         Some(init_order.clone()),
                         &TENANTS,
                         &ctx,
@@ -161,7 +162,8 @@ pub async fn init_tenant_mgr(
 pub(crate) fn schedule_local_tenant_processing(
     conf: &'static PageServerConf,
     tenant_path: &Path,
-    resources: TenantSharedResources,
+    broker_client: storage_broker::BrokerClientChannel,
+    remote_storage: Option<GenericRemoteStorage>,
     init_order: Option<InitializationOrder>,
     tenants: &'static tokio::sync::RwLock<TenantsMap>,
     ctx: &RequestContext,
@@ -198,15 +200,9 @@ pub(crate) fn schedule_local_tenant_processing(
 
     let tenant = if conf.tenant_attaching_mark_file_path(&tenant_id).exists() {
         info!("tenant {tenant_id} has attaching mark file, resuming its attach operation");
-        if let Some(remote_storage) = resources.remote_storage {
-            match Tenant::spawn_attach(
-                conf,
-                tenant_id,
-                resources.broker_client,
-                tenants,
-                remote_storage,
-                ctx,
-            ) {
+        if let Some(remote_storage) = remote_storage {
+            match Tenant::spawn_attach(conf, tenant_id, broker_client, tenants, remote_storage, ctx)
+            {
                 Ok(tenant) => tenant,
                 Err(e) => {
                     error!("Failed to spawn_attach tenant {tenant_id}, reason: {e:#}");
@@ -224,7 +220,15 @@ pub(crate) fn schedule_local_tenant_processing(
     } else {
         info!("tenant {tenant_id} is assumed to be loadable, starting load operation");
         // Start loading the tenant into memory. It will initially be in Loading state.
-        Tenant::spawn_load(conf, tenant_id, resources, init_order, tenants, ctx)
+        Tenant::spawn_load(
+            conf,
+            tenant_id,
+            broker_client,
+            remote_storage,
+            init_order,
+            tenants,
+            ctx,
+        )
     };
     Ok(tenant)
 }
@@ -359,12 +363,8 @@ pub async fn create_tenant(
         // TODO: tenant directory remains on disk if we bail out from here on.
         //       See https://github.com/neondatabase/neon/issues/4233
 
-        let tenant_resources = TenantSharedResources {
-            broker_client,
-            remote_storage,
-        };
         let created_tenant =
-            schedule_local_tenant_processing(conf, &tenant_directory, tenant_resources, None, &TENANTS, ctx)?;
+            schedule_local_tenant_processing(conf, &tenant_directory, broker_client, remote_storage, None, &TENANTS, ctx)?;
         // TODO: tenant object & its background loops remain, untracked in tenant map, if we fail here.
         //      See https://github.com/neondatabase/neon/issues/4233
 
@@ -523,11 +523,7 @@ pub async fn load_tenant(
                 .with_context(|| format!("Failed to remove tenant ignore mark {tenant_ignore_mark:?} during tenant loading"))?;
         }
 
-        let resources = TenantSharedResources {
-            broker_client,
-            remote_storage,
-        };
-        let new_tenant = schedule_local_tenant_processing(conf, &tenant_path,  resources, None,  &TENANTS, ctx)
+        let new_tenant = schedule_local_tenant_processing(conf, &tenant_path, broker_client, remote_storage, None, &TENANTS, ctx)
             .with_context(|| {
                 format!("Failed to schedule tenant processing in path {tenant_path:?}")
             })?;
@@ -608,11 +604,7 @@ pub async fn attach_tenant(
             .context("check for attach marker file existence")?;
         anyhow::ensure!(marker_file_exists, "create_tenant_files should have created the attach marker file");
 
-        let resources = TenantSharedResources {
-            broker_client,
-            remote_storage: Some(remote_storage),
-        };
-        let attached_tenant = schedule_local_tenant_processing(conf, &tenant_dir, resources, None, &TENANTS, ctx)?;
+        let attached_tenant = schedule_local_tenant_processing(conf, &tenant_dir, broker_client, Some(remote_storage), None, &TENANTS, ctx)?;
         // TODO: tenant object & its background loops remain, untracked in tenant map, if we fail here.
         //      See https://github.com/neondatabase/neon/issues/4233
 

pageserver/src/tenant/remote_timeline_client.rs

@@ -1578,11 +1578,7 @@ mod tests {
         };
 
         assert_file_list(
-            &index_part
-                .layer_metadata
-                .keys()
-                .map(|f| f.to_owned())
-                .collect(),
+            &index_part.timeline_layers,
             &[
                 &layer_file_name_1.file_name(),
                 &layer_file_name_2.file_name(),

pageserver/src/tenant/remote_timeline_client/index.rs

@@ -62,9 +62,10 @@ pub struct IndexPart {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub deleted_at: Option<NaiveDateTime>,
 
-    /// Legacy field: equal to the keys of `layer_metadata`, only written out for forward compat
-    #[serde(default, skip_deserializing)]
-    timeline_layers: HashSet<LayerFileName>,
+    /// Layer names, which are stored on the remote storage.
+    ///
+    /// Additional metadata can might exist in `layer_metadata`.
+    pub timeline_layers: HashSet<LayerFileName>,
 
     /// Per layer file name metadata, which can be present for a present or missing layer file.
     ///
@@ -73,10 +74,9 @@ pub struct IndexPart {
     pub layer_metadata: HashMap<LayerFileName, IndexLayerMetadata>,
 
     // 'disk_consistent_lsn' is a copy of the 'disk_consistent_lsn' in the metadata.
-    // It's duplicated for convenience when reading the serialized structure, but is
-    // private because internally we would read from metadata instead.
+    // It's duplicated here for convenience.
     #[serde_as(as = "DisplayFromStr")]
-    disk_consistent_lsn: Lsn,
+    pub disk_consistent_lsn: Lsn,
     metadata_bytes: Vec<u8>,
 }
 
@@ -85,11 +85,7 @@ impl IndexPart {
     /// used to understand later versions.
     ///
     /// Version is currently informative only.
-    /// Version history
-    /// - 2: added `deleted_at`
-    /// - 3: no longer deserialize `timeline_layers` (serialized format is the same, but timeline_layers
-    ///      is always generated from the keys of `layer_metadata`)
-    const LATEST_VERSION: usize = 3;
+    const LATEST_VERSION: usize = 2;
     pub const FILE_NAME: &'static str = "index_part.json";
 
     pub fn new(
@@ -170,7 +166,7 @@ mod tests {
         let expected = IndexPart {
             // note this is not verified, could be anything, but exists for humans debugging.. could be the git version instead?
             version: 1,
-            timeline_layers: HashSet::new(),
+            timeline_layers: HashSet::from(["000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__0000000001696070-00000000016960E9".parse().unwrap()]),
             layer_metadata: HashMap::from([
                 ("000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__0000000001696070-00000000016960E9".parse().unwrap(), IndexLayerMetadata {
                     file_size: 25600000,
@@ -207,7 +203,7 @@ mod tests {
         let expected = IndexPart {
             // note this is not verified, could be anything, but exists for humans debugging.. could be the git version instead?
             version: 1,
-            timeline_layers: HashSet::new(),
+            timeline_layers: HashSet::from(["000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__0000000001696070-00000000016960E9".parse().unwrap()]),
             layer_metadata: HashMap::from([
                 ("000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__0000000001696070-00000000016960E9".parse().unwrap(), IndexLayerMetadata {
                     file_size: 25600000,
@@ -245,7 +241,7 @@ mod tests {
         let expected = IndexPart {
             // note this is not verified, could be anything, but exists for humans debugging.. could be the git version instead?
             version: 2,
-            timeline_layers: HashSet::new(),
+            timeline_layers: HashSet::from(["000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__0000000001696070-00000000016960E9".parse().unwrap()]),
             layer_metadata: HashMap::from([
                 ("000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__0000000001696070-00000000016960E9".parse().unwrap(), IndexLayerMetadata {
                     file_size: 25600000,

pageserver/src/tenant/timeline.rs

@@ -140,12 +140,6 @@ fn drop_rlock<T>(rlock: tokio::sync::OwnedRwLockReadGuard<T>) {
 fn drop_wlock<T>(rlock: tokio::sync::RwLockWriteGuard<'_, T>) {
     drop(rlock)
 }
-
-/// The outward-facing resources required to build a Timeline
-pub struct TimelineResources {
-    pub remote_client: Option<RemoteTimelineClient>,
-}
-
 pub struct Timeline {
     conf: &'static PageServerConf,
     tenant_conf: Arc<RwLock<TenantConfOpt>>,
@@ -1380,7 +1374,7 @@ impl Timeline {
         timeline_id: TimelineId,
         tenant_id: TenantId,
         walredo_mgr: Arc<dyn WalRedoManager + Send + Sync>,
-        resources: TimelineResources,
+        remote_client: Option<RemoteTimelineClient>,
         pg_version: u32,
         initial_logical_size_can_start: Option<completion::Barrier>,
         initial_logical_size_attempt: Option<completion::Completion>,
@@ -1415,7 +1409,7 @@ impl Timeline {
                 walredo_mgr,
                 walreceiver: Mutex::new(None),
 
-                remote_client: resources.remote_client.map(Arc::new),
+                remote_client: remote_client.map(Arc::new),
 
                 // initialize in-memory 'last_record_lsn' from 'disk_consistent_lsn'.
                 last_record_lsn: SeqWait::new(RecordLsn {
@@ -1736,7 +1730,7 @@ impl Timeline {
 
         let mut corrupted_local_layers = Vec::new();
         let mut added_remote_layers = Vec::new();
-        for remote_layer_name in index_part.layer_metadata.keys() {
+        for remote_layer_name in &index_part.timeline_layers {
             let local_layer = local_only_layers.remove(remote_layer_name);
 
             let remote_layer_metadata = index_part
@@ -1896,7 +1890,7 @@ impl Timeline {
             Some(index_part) => {
                 info!(
                     "initializing upload queue from remote index with {} layer files",
-                    index_part.layer_metadata.len()
+                    index_part.timeline_layers.len()
                 );
                 remote_client.init_upload_queue(index_part)?;
                 self.create_remote_layers(index_part, local_layers, disk_consistent_lsn)

pageserver/src/tenant/timeline/delete.rs

@@ -25,7 +25,7 @@ use crate::{
     InitializationOrder,
 };
 
-use super::{Timeline, TimelineResources};
+use super::Timeline;
 
 /// Now that the Timeline is in Stopping state, request all the related tasks to shut down.
 async fn stop_tasks(timeline: &Timeline) -> Result<(), DeleteTimelineError> {
@@ -416,7 +416,7 @@ impl DeleteTimelineFlow {
                 timeline_id,
                 local_metadata,
                 None, // Ancestor is not needed for deletion.
-                TimelineResources { remote_client },
+                remote_client,
                 init_order,
                 // Important. We dont pass ancestor above because it can be missing.
                 // Thus we need to skip the validation here.

pageserver/src/tenant/upload_queue.rs

@@ -140,12 +140,23 @@ impl UploadQueue {
             }
         }
 
-        let mut files = HashMap::with_capacity(index_part.layer_metadata.len());
-        for (layer_name, layer_metadata) in &index_part.layer_metadata {
-            files.insert(
-                layer_name.to_owned(),
-                LayerFileMetadata::from(layer_metadata),
-            );
+        let mut files = HashMap::with_capacity(index_part.timeline_layers.len());
+        for layer_name in &index_part.timeline_layers {
+            match index_part
+                .layer_metadata
+                .get(layer_name)
+                .map(LayerFileMetadata::from)
+            {
+                Some(layer_metadata) => {
+                    files.insert(layer_name.to_owned(), layer_metadata);
+                }
+                None => {
+                    anyhow::bail!(
+                        "No remote layer metadata found for layer {}",
+                        layer_name.file_name()
+                    );
+                }
+            }
         }
 
         let index_part_metadata = index_part.parse_metadata()?;

pgxn/neon/walproposer.c

@@ -163,6 +163,7 @@ static void nwp_register_gucs(void);
 static void nwp_prepare_shmem(void);
 static uint64 backpressure_lag_impl(void);
 static bool backpressure_throttling_impl(void);
+static uint64 measure_replication_lag(void); 
 
 static process_interrupts_callback_t PrevProcessInterruptsCallback;
 static shmem_startup_hook_type prev_shmem_startup_hook_type;
@@ -171,6 +172,8 @@ static shmem_request_hook_type prev_shmem_request_hook = NULL;
 static void walproposer_shmem_request(void);
 #endif
 
+static bool check_replication_lag;
+
 void
 pg_init_walproposer(void)
 {
@@ -2492,37 +2495,45 @@ backpressure_lag_impl(void)
 {
 	if (max_replication_apply_lag > 0 || max_replication_flush_lag > 0 || max_replication_write_lag > 0)
 	{
-		XLogRecPtr	writePtr;
-		XLogRecPtr	flushPtr;
-		XLogRecPtr	applyPtr;
+		check_replication_lag = true;
+		return measure_replication_lag();
+	}
+	return 0;
+}
+
+static uint64
+measure_replication_lag(void)
+{
+	XLogRecPtr	writePtr;
+	XLogRecPtr	flushPtr;
+	XLogRecPtr	applyPtr;
 #if PG_VERSION_NUM >= 150000
-		XLogRecPtr	myFlushLsn = GetFlushRecPtr(NULL);
+	XLogRecPtr	myFlushLsn = GetFlushRecPtr(NULL);
 #else
-		XLogRecPtr	myFlushLsn = GetFlushRecPtr();
+	XLogRecPtr	myFlushLsn = GetFlushRecPtr();
 #endif
-		replication_feedback_get_lsns(&writePtr, &flushPtr, &applyPtr);
+	replication_feedback_get_lsns(&writePtr, &flushPtr, &applyPtr);
 #define MB ((XLogRecPtr)1024 * 1024)
 
-		elog(DEBUG2, "current flushLsn %X/%X PageserverFeedback: write %X/%X flush %X/%X apply %X/%X",
-			 LSN_FORMAT_ARGS(myFlushLsn),
-			 LSN_FORMAT_ARGS(writePtr),
-			 LSN_FORMAT_ARGS(flushPtr),
-			 LSN_FORMAT_ARGS(applyPtr));
+	elog(DEBUG2, "current flushLsn %X/%X PageserverFeedback: write %X/%X flush %X/%X apply %X/%X",
+		 LSN_FORMAT_ARGS(myFlushLsn),
+		 LSN_FORMAT_ARGS(writePtr),
+		 LSN_FORMAT_ARGS(flushPtr),
+		 LSN_FORMAT_ARGS(applyPtr));
 
-		if ((writePtr != InvalidXLogRecPtr && max_replication_write_lag > 0 && myFlushLsn > writePtr + max_replication_write_lag * MB))
-		{
-			return (myFlushLsn - writePtr - max_replication_write_lag * MB);
-		}
+	if ((writePtr != InvalidXLogRecPtr && max_replication_write_lag > 0 && myFlushLsn > writePtr + max_replication_write_lag * MB))
+	{
+		return (myFlushLsn - writePtr - max_replication_write_lag * MB);
+	}
 
-		if ((flushPtr != InvalidXLogRecPtr && max_replication_flush_lag > 0 && myFlushLsn > flushPtr + max_replication_flush_lag * MB))
-		{
-			return (myFlushLsn - flushPtr - max_replication_flush_lag * MB);
-		}
+	if ((flushPtr != InvalidXLogRecPtr && max_replication_flush_lag > 0 && myFlushLsn > flushPtr + max_replication_flush_lag * MB))
+	{
+		return (myFlushLsn - flushPtr - max_replication_flush_lag * MB);
+	}
 
-		if ((applyPtr != InvalidXLogRecPtr && max_replication_apply_lag > 0 && myFlushLsn > applyPtr + max_replication_apply_lag * MB))
-		{
-			return (myFlushLsn - applyPtr - max_replication_apply_lag * MB);
-		}
+	if ((applyPtr != InvalidXLogRecPtr && max_replication_apply_lag > 0 && myFlushLsn > applyPtr + max_replication_apply_lag * MB))
+	{
+		return (myFlushLsn - applyPtr - max_replication_apply_lag * MB);
 	}
 	return 0;
 }
@@ -2539,14 +2550,18 @@ backpressure_throttling_impl(void)
 	? PrevProcessInterruptsCallback()
 	: false;
 
-	/* Don't throttle read only transactions and wal sender. */
-	if (am_walsender || !TransactionIdIsValid(GetCurrentTransactionIdIfAny()))
+	/* Throttle onlhy backends writing WAL. */
+	if (!check_replication_lag)
 		return retry;
 
-	/* Calculate replicas lag */
-	lag = backpressure_lag_impl();
+	/* Calculate replication lag */
+	lag = measure_replication_lag();
 	if (lag == 0)
+	{
+		/* Do not measure replication lag before we writting something to the WAL */
+		check_replication_lag = false;
 		return retry;
+	}
 
 	/* Suspend writers until replicas catch up */
 	set_ps_display("backpressure throttling");

proxy/src/http/conn_pool.rs

@@ -64,13 +64,13 @@ pub struct EndpointConnPool {
     total_conns: usize,
 }
 
-/// 4096 is the number of rounds that SCRAM-SHA-256 recommends.
-/// It's not the 600,000 that OWASP recommends... but our passwords are high entropy anyway.
+/// This is cheap and not hugely secure.
+/// But probably good enough for in memory only hashes.
 ///
-/// Still takes 1.4ms to hash on my hardware.
+/// Still takes 3.5ms to hash on my hardware.
 /// We don't want to ruin the latency improvements of using the pool by making password verification take too long
 const PARAMS: Params = Params {
-    rounds: 4096,
+    rounds: 10_000,
     output_length: 32,
 };
 
@@ -99,10 +99,6 @@ pub struct GlobalConnPool {
     max_conns_per_endpoint: usize,
 
     proxy_config: &'static crate::config::ProxyConfig,
-
-    // Using a lock to remove any race conditions.
-    // Eg cleaning up connections while a new connection is returned
-    closed: RwLock<bool>,
 }
 
 impl GlobalConnPool {
@@ -112,24 +108,9 @@ impl GlobalConnPool {
             global_pool_size: AtomicUsize::new(0),
             max_conns_per_endpoint: MAX_CONNS_PER_ENDPOINT,
             proxy_config: config,
-            closed: RwLock::new(false),
         })
     }
 
-    pub fn shutdown(&self) {
-        *self.closed.write() = true;
-
-        self.global_pool.retain(|_, endpoint_pool| {
-            let mut pool = endpoint_pool.write();
-            // by clearing this hashmap, we remove the slots that a connection can be returned to.
-            // when returning, it drops the connection if the slot doesn't exist
-            pool.pools.clear();
-            pool.total_conns = 0;
-
-            false
-        });
-    }
-
     pub async fn get(
         &self,
         conn_info: &ConnInfo,
@@ -227,20 +208,7 @@ impl GlobalConnPool {
         new_client
     }
 
-    pub fn put(&self, conn_info: &ConnInfo, client: Client) -> anyhow::Result<()> {
-        // We want to hold this open while we return. This ensures that the pool can't close
-        // while we are in the middle of returning the connection.
-        let closed = self.closed.read();
-        if *closed {
-            info!("pool: throwing away connection '{conn_info}' because pool is closed");
-            return Ok(());
-        }
-
-        if client.inner.is_closed() {
-            info!("pool: throwing away connection '{conn_info}' because connection is closed");
-            return Ok(());
-        }
-
+    pub async fn put(&self, conn_info: &ConnInfo, client: Client) -> anyhow::Result<()> {
         let pool = self.get_or_create_endpoint_pool(&conn_info.hostname);
 
         // return connection to the pool
@@ -408,9 +376,9 @@ async fn connect_to_compute_once(
     let (tx, mut rx) = tokio::sync::watch::channel(session);
 
     let conn_id = uuid::Uuid::new_v4();
-    let span = info_span!(parent: None, "connection", %conn_id);
+    let span = info_span!(parent: None, "connection", %conn_info, %conn_id);
     span.in_scope(|| {
-        info!(%conn_info, %session, "new connection");
+        info!(%session, "new connection");
     });
 
     tokio::spawn(
@@ -420,28 +388,26 @@ async fn connect_to_compute_once(
                 info!(%session, "changed session");
             }
 
-            loop {
-                let message = ready!(connection.poll_message(cx));
+            let message = ready!(connection.poll_message(cx));
 
-                match message {
-                    Some(Ok(AsyncMessage::Notice(notice))) => {
-                        info!(%session, "notice: {}", notice);
-                    }
-                    Some(Ok(AsyncMessage::Notification(notif))) => {
-                        warn!(%session, pid = notif.process_id(), channel = notif.channel(), "notification received");
-                    }
-                    Some(Ok(_)) => {
-                        warn!(%session, "unknown message");
-                    }
-                    Some(Err(e)) => {
-                        error!(%session, "connection error: {}", e);
-                        return Poll::Ready(())
-                    }
-                    None => {
-                        info!("connection closed");
-                        return Poll::Ready(())
-                    }
+            match message {
+                Some(Ok(AsyncMessage::Notice(notice))) => {
+                    info!(%session, "notice: {}", notice);
+                    Poll::Pending
                 }
+                Some(Ok(AsyncMessage::Notification(notif))) => {
+                    warn!(%session, pid = notif.process_id(), channel = notif.channel(), "notification received");
+                    Poll::Pending
+                }
+                Some(Ok(_)) => {
+                    warn!(%session, "unknown message");
+                    Poll::Pending
+                }
+                Some(Err(e)) => {
+                    error!(%session, "connection error: {}", e);
+                    Poll::Ready(())
+                }
+                None => Poll::Ready(()),
             }
         })
         .instrument(span)

proxy/src/http/sql_over_http.rs

@@ -16,6 +16,7 @@ use tokio_postgres::types::Type;
 use tokio_postgres::GenericClient;
 use tokio_postgres::IsolationLevel;
 use tokio_postgres::Row;
+use tracing::Instrument;
 use url::Url;
 
 use super::conn_pool::ConnInfo;
@@ -285,12 +286,13 @@ pub async fn handle(
     };
 
     if allow_pool {
-        let current_span = tracing::Span::current();
         // return connection to the pool
-        tokio::task::spawn_blocking(move || {
-            let _span = current_span.enter();
-            let _ = conn_pool.put(&conn_info, client);
-        });
+        tokio::task::spawn(
+            async move {
+                let _ = conn_pool.put(&conn_info, client).await;
+            }
+            .in_current_span(),
+        );
     }
 
     result

proxy/src/http/websocket.rs

@@ -269,18 +269,6 @@ pub async fn task_main(
 
     let conn_pool: Arc<GlobalConnPool> = GlobalConnPool::new(config);
 
-    // shutdown the connection pool
-    tokio::spawn({
-        let cancellation_token = cancellation_token.clone();
-        let conn_pool = conn_pool.clone();
-        async move {
-            cancellation_token.cancelled().await;
-            tokio::task::spawn_blocking(move || conn_pool.shutdown())
-                .await
-                .unwrap();
-        }
-    });
-
     let tls_config = config.tls_config.as_ref().map(|cfg| cfg.to_server_config());
     let tls_acceptor: tokio_rustls::TlsAcceptor = match tls_config {
         Some(config) => config.into(),