Merge commit 'd96cea191' into problame/standby-horizon-leases

Heikki implemented a better representation of pageserver conn info which
obsoletes my refactoring attempt. Use it.

Cargo.lock

@@ -4652,6 +4652,7 @@ version = "0.1.0"
 dependencies = [
  "anyhow",
  "bytes",
+ "chrono",
  "futures",
  "pageserver_api",
  "postgres_ffi_types",

compute_tools/src/compute.rs

@@ -52,7 +52,6 @@ use crate::logger::{self, startup_context_from_env};
 use crate::lsn_lease::launch_lsn_lease_bg_task_for_static;
 use crate::metrics::COMPUTE_CTL_UP;
 use crate::monitor::launch_monitor;
-use crate::pg_helpers::*;
 use crate::pgbouncer::*;
 use crate::rsyslog::{
     PostgresLogsRsyslogConfig, configure_audit_rsyslog, configure_postgres_logs_export,
@@ -63,6 +62,7 @@ use crate::swap::resize_swap;
 use crate::sync_sk::{check_if_synced, ping_safekeeper};
 use crate::tls::watch_cert_for_changes;
 use crate::{config, extension_server, local_proxy};
+use crate::{pg_helpers::*, ro_replica};
 
 pub static SYNC_SAFEKEEPERS_PID: AtomicU32 = AtomicU32::new(0);
 pub static PG_PID: AtomicU32 = AtomicU32::new(0);
@@ -156,6 +156,8 @@ pub struct ComputeNode {
     pub ext_download_progress: RwLock<HashMap<String, (DateTime<Utc>, bool)>>,
     pub compute_ctl_config: ComputeCtlConfig,
 
+    pub(crate) ro_replica: Arc<ro_replica::GlobalState>,
+
     /// Handle to the extension stats collection task
     extension_stats_task: TaskHandle,
     lfc_offload_task: TaskHandle,
@@ -619,6 +621,7 @@ impl ComputeNode {
             compute_ctl_config: config.compute_ctl_config,
             extension_stats_task: Mutex::new(None),
             lfc_offload_task: Mutex::new(None),
+            ro_replica: Arc::new(ro_replica::GlobalState::default()),
         })
     }
 
@@ -672,6 +675,8 @@ impl ComputeNode {
 
         launch_lsn_lease_bg_task_for_static(&this);
 
+        ro_replica::spawn_bg_task(Arc::clone(&this));
+
         // We have a spec, start the compute
         let mut delay_exit = false;
         let mut vm_monitor = None;

compute_tools/src/lib.rs

@@ -27,6 +27,7 @@ pub mod params;
 pub mod pg_helpers;
 pub mod pg_isready;
 pub mod pgbouncer;
+pub(crate) mod ro_replica;
 pub mod rsyslog;
 pub mod spec;
 mod spec_apply;

compute_tools/src/monitor.rs

@@ -4,9 +4,10 @@ use std::time::Duration;
 
 use chrono::{DateTime, Utc};
 use compute_api::responses::ComputeStatus;
-use compute_api::spec::ComputeFeature;
+use compute_api::spec::{ComputeFeature, ComputeMode};
 use postgres::{Client, NoTls};
 use tracing::{Level, error, info, instrument, span};
+use utils::lsn::Lsn;
 
 use crate::compute::ComputeNode;
 use crate::metrics::{PG_CURR_DOWNTIME_MS, PG_TOTAL_DOWNTIME_MS};
@@ -346,6 +347,47 @@ impl ComputeMonitor {
             }
         }
 
+        if self
+            .compute
+            .has_feature(ComputeFeature::StandbyHorizonLeasesExperimental)
+        {
+            let mode: ComputeMode = self
+                .compute
+                .state
+                .lock()
+                .unwrap()
+                .pspec
+                .as_ref()
+                .expect("we launch ComputeMonitor only after we received a spec")
+                .spec
+                .mode;
+            match mode {
+                // TODO: can the .spec.mode ever change? if it can (e.g. secondary promote to primary)
+                // then we should make sure that lsn_lease_state transitions back to None so we stop renewing it.
+                ComputeMode::Primary => (),
+                ComputeMode::Static(_) => (),
+                ComputeMode::Replica => {
+                    // TODO: instead of apply_lsn, use min inflight request LSN
+                    match cli.query_one("SELECT pg_last_wal_replay_lsn() as apply_lsn", &[]) {
+                        Ok(r) => match r.try_get::<&str, postgres_types::PgLsn>("apply_lsn") {
+                            Ok(apply_lsn) => {
+                                let apply_lsn = Lsn(apply_lsn.into());
+                                self.compute
+                                    .ro_replica
+                                    .update_min_inflight_request_lsn(apply_lsn);
+                            }
+                            Err(e) => {
+                                anyhow::bail!("parse apply_lsn: {e}");
+                            }
+                        },
+                        Err(e) => {
+                            anyhow::bail!("query apply_lsn: {e}");
+                        }
+                    }
+                }
+            }
+        }
+
         Ok(())
     }
 }

compute_tools/src/ro_replica.rs

@@ -0,0 +1,320 @@
+use std::{str::FromStr, sync::Arc, time::Duration};
+
+use anyhow::Context;
+use chrono::Utc;
+use compute_api::spec::PageserverProtocol;
+use futures::{FutureExt, StreamExt, stream::FuturesUnordered};
+use postgres::SimpleQueryMessage;
+use tokio_util::sync::CancellationToken;
+use tracing::{Instrument, error, info, info_span, instrument, warn};
+use utils::{backoff::retry, id::TimelineId, lsn::Lsn, shard::TenantShardId};
+
+use crate::compute::ComputeNode;
+
+#[derive(Default)]
+pub(crate) struct GlobalState {
+    min_inflight_request_lsn: tokio::sync::watch::Sender<Option<Lsn>>,
+}
+
+impl GlobalState {
+    pub fn update_min_inflight_request_lsn(&self, update: Lsn) {
+        self.min_inflight_request_lsn.send_if_modified(|value| {
+            let modified = *value != Some(update);
+            if let Some(value) = *value && value > update {
+                warn!(current=%value, new=%update, "min inflight request lsn moving backwards, this should not happen, bug in communicator");
+            }
+            *value = Some(update);
+            modified
+        });
+    }
+}
+
+pub fn spawn_bg_task(compute: Arc<ComputeNode>) {
+    std::thread::spawn(|| {
+        let rt = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+        rt.block_on(bg_task(compute))
+    });
+}
+
+#[instrument(name = "standby_horizon_lease", skip_all, fields(lease_id))]
+async fn bg_task(compute: Arc<ComputeNode>) {
+    // Use a lease_id that is globally unique to this process to maximize attribution precision & log correlation.
+    let lease_id = format!("v1-{}-{}", compute.params.compute_id, std::process::id());
+    tracing::Span::current().record("lease_id", tracing::field::display(&lease_id));
+
+    // XXX can we get tenant_id and timeline_id now so they're included in log span?
+    // These should be immutable but it requires pspec is set alreayd. Is it?
+
+    // Wait until we have the first value.
+    // Allows us to simply .unwrap() later because it never transitions back to None.
+    // XXX maybe we should transition back to None and use that to detect replica promotion (where we stop renewing the lease?)
+    info!("waiting for first lease lsn to be fetched from postgres");
+    let mut min_inflight_request_lsn_changed =
+        compute.ro_replica.min_inflight_request_lsn.subscribe();
+    min_inflight_request_lsn_changed.mark_changed(); // it could have been set already
+    min_inflight_request_lsn_changed
+        .wait_for(|value| value.is_some())
+        .await
+        .expect("we never drop the sender");
+
+    // React to connstring changes. Sadly there is no async API for this yet.
+    let (connstr_watch_tx, mut connstr_watch_rx) = tokio::sync::watch::channel(None);
+    std::thread::spawn({
+        let compute = Arc::clone(&compute);
+        move || {
+            loop {
+                compute.wait_timeout_while_pageserver_connstr_unchanged(Duration::MAX);
+                let new = compute
+                    .state
+                    .lock()
+                    .unwrap()
+                    .pspec
+                    .as_ref()
+                    .and_then(|pspec| pspec.spec.pageserver_connstring.clone());
+                connstr_watch_tx.send_if_modified(|existing| {
+                    if &new != existing {
+                        *existing = new;
+                        true
+                    } else {
+                        false
+                    }
+                });
+            }
+        }
+    });
+
+    let mut obtained = ObtainedLease {
+        lsn: Lsn(0),
+        nearest_expiration: Utc::now(),
+    };
+    loop {
+        let valid_duration: Duration = obtained
+            .nearest_expiration
+            .signed_duration_since(Utc::now())
+            .to_std()
+            // to_std() errors if the duration is less than zero, i.e,. if the lease already expired;
+            // try to renew anyway in that case;
+            .unwrap_or_default();
+        // Sleep for 60 seconds less than the valid duration but no more than half of the valid duration.
+        let sleep_duration = valid_duration
+            .saturating_sub(Duration::from_secs(60))
+            .max(valid_duration / 2);
+
+        tokio::select! {
+            _  = tokio::time::sleep(sleep_duration) => {
+                info!("updating because lease is going to expire soon");
+            }
+            _ = connstr_watch_rx.changed() => {
+                info!("updating due to changed pageserver_connstr")
+            }
+            _ = async {
+                // debounce; TODO make this lower in tests
+                tokio::time::sleep(Duration::from_secs(10)).await;
+                // every 10 GiB; TODO make this tighter in tests?
+                let max_horizon_lag = 10 * (1<<30);
+                min_inflight_request_lsn_changed.wait_for(|x| x.unwrap().0 > obtained.lsn.0 + max_horizon_lag).await
+            } => {
+                info!(%obtained.lsn, "updating due to max horizon lag");
+            }
+        }
+        // retry forever
+        let compute = Arc::clone(&compute);
+        let lease_id = lease_id.clone();
+        obtained = retry(
+            || attempt(lease_id.clone(), &compute),
+            |_| false,
+            0,
+            u32::MAX, // forever
+            "update standby_horizon position in pageserver",
+            // There is no cancellation story in compute_ctl
+            &CancellationToken::new(),
+        )
+        .await
+        .expect("is_permanent returns false, so, retry always returns Some")
+        .expect("u32::MAX exceeded");
+    }
+}
+
+struct ObtainedLease {
+    lsn: Lsn,
+    nearest_expiration: chrono::DateTime<Utc>,
+}
+
+async fn attempt(lease_id: String, compute: &Arc<ComputeNode>) -> anyhow::Result<ObtainedLease> {
+    // Note: List of pageservers is dynamic, need to re-read configs before each attempt.
+    let (conninfo, auth, tenant_id, timeline_id) = {
+        let state = compute.state.lock().unwrap();
+        let pspec = state.pspec.as_ref().expect("spec must be set");
+        (
+            pspec.pageserver_conninfo.clone(),
+            pspec.storage_auth_token.clone(),
+            pspec.tenant_id,
+            pspec.timeline_id,
+        )
+    };
+
+    let lsn = compute
+        .ro_replica
+        .min_inflight_request_lsn
+        .borrow()
+        .expect("we only call this function once it has been transitioned to Some");
+
+    let mut futs = FuturesUnordered::new();
+    for (shard_index, shard) in conninfo.shards {
+        let tenant_shard_id = TenantShardId {
+            tenant_id,
+            shard_number: shard_index.shard_number,
+            shard_count: shard_index.shard_count,
+        };
+        // XXX: If there are more than pageserver for the one shard, do we need to get a
+        // leas on all of them? Currently, that's what we assume, but this is hypothetical
+        // as of this writing, as we never pass the info for more than one pageserver per
+        // shard.
+        for pageserver in shard.pageservers {
+            let logging_span = info_span!(
+                "attempt_one",
+                tenant_id=%tenant_shard_id.tenant_id,
+                shard_id=%tenant_shard_id.shard_slug(),
+                timeline_id=%timeline_id,
+                pageserver_id=?pageserver.id,
+                protocol=?conninfo.prefer_protocol,
+            );
+            let logging_wrapper = async |fut| {
+                async move {
+                    // TODO: timeout?
+                    match fut.await {
+                        Ok(Some(v)) => {
+                            info!("lease obtained");
+                            Ok(Some(v))
+                        }
+                        Ok(None) => {
+                            error!("pageserver rejected our request");
+                            Ok(None)
+                        }
+                        Err(err) => {
+                            error!("communication failure: {err:?}");
+                            Err(())
+                        }
+                    }
+                }
+                .instrument(logging_span)
+                .await
+            };
+
+            let fut = match conninfo.prefer_protocol {
+                PageserverProtocol::Libpq => logging_wrapper(
+                    attempt_one_libpq(
+                        pageserver.libpq_url.clone().unwrap(),
+                        auth.clone(),
+                        tenant_shard_id,
+                        timeline_id,
+                        lease_id.clone(),
+                        lsn,
+                    )
+                    .boxed(),
+                ),
+                PageserverProtocol::Grpc => logging_wrapper(
+                    attempt_one_grpc(
+                        pageserver.grpc_url.clone().unwrap(),
+                        auth.clone(),
+                        tenant_shard_id,
+                        timeline_id,
+                        lease_id.clone(),
+                        lsn,
+                    )
+                    .boxed(),
+                ),
+            };
+            futs.push(fut);
+        }
+    }
+    let mut errors = 0;
+    let mut nearest_expiration = None;
+    while let Some(res) = futs.next().await {
+        match res {
+            Ok(Some(expiration)) => {
+                let nearest_expiration = nearest_expiration.get_or_insert(expiration);
+                *nearest_expiration = std::cmp::min(*nearest_expiration, expiration);
+            }
+            Ok(None) | Err(()) => {
+                // the logging wrapper does the logging
+                errors += 1;
+            }
+        }
+    }
+    if errors > 0 {
+        return Err(anyhow::anyhow!(
+            "failed to advance standby_horizon for {errors} shards, check logs for details"
+        ));
+    }
+    match nearest_expiration {
+        Some(nearest_expiration) => Ok(ObtainedLease {
+            lsn,
+            nearest_expiration,
+        }),
+        None => Err(anyhow::anyhow!("pageservers connstrings is empty")), // this probably can't happen
+    }
+}
+
+async fn attempt_one_libpq(
+    connstring: String,
+    auth: Option<String>,
+    tenant_shard_id: TenantShardId,
+    timeline_id: TimelineId,
+    lease_id: String,
+    lsn: Lsn,
+) -> anyhow::Result<Option<chrono::DateTime<Utc>>> {
+    let mut config = tokio_postgres::Config::from_str(&connstring)?;
+    if let Some(auth) = auth {
+        config.password(auth);
+    }
+    let (client, conn) = config.connect(postgres::NoTls).await?;
+    tokio::spawn(conn);
+    let cmd = format!("lease standby_horizon {tenant_shard_id} {timeline_id} {lease_id} {lsn} ");
+    let res = client.simple_query(&cmd).await?;
+    let msg = match res.first() {
+        Some(msg) => msg,
+        None => anyhow::bail!("empty response"),
+    };
+    let row = match msg {
+        SimpleQueryMessage::Row(row) => row,
+        _ => anyhow::bail!("expected row message type"),
+    };
+
+    // Note: this will be NULL (=> None) if a lease is explicitly not granted.
+    row.get("expiration")
+        .map(|s| {
+            chrono::DateTime::<Utc>::from_str(s).with_context(|| format!("parse expiration: {s:?}"))
+        })
+        .transpose()
+}
+
+async fn attempt_one_grpc(
+    connstring: String,
+    auth: Option<String>,
+    tenant_shard_id: TenantShardId,
+    timeline_id: TimelineId,
+    lease_id: String,
+    lsn: Lsn,
+) -> anyhow::Result<Option<chrono::DateTime<Utc>>> {
+    let mut client = pageserver_page_api::Client::connect(
+        connstring,
+        tenant_shard_id.tenant_id,
+        timeline_id,
+        tenant_shard_id.to_index(),
+        auth,
+        None,
+    )
+    .await?;
+
+    let req = pageserver_page_api::LeaseStandbyHorizonRequest { lease_id, lsn };
+    match client.lease_standby_horizon(req).await {
+        Ok(pageserver_page_api::LeaseStandbyHorizonResponse { expiration }) => Ok(Some(expiration)),
+        // Lease couldn't be acquired
+        Err(err) if err.code() == tonic::Code::FailedPrecondition => Ok(None),
+        Err(err) => Err(err.into()),
+    }
+}

control_plane/src/bin/neon_local.rs

@@ -16,7 +16,7 @@ use std::time::Duration;
 use anyhow::{Context, Result, anyhow, bail};
 use clap::Parser;
 use compute_api::requests::ComputeClaimsScope;
-use compute_api::spec::{ComputeMode, PageserverProtocol};
+use compute_api::spec::{ComputeFeature, ComputeMode, PageserverProtocol};
 use control_plane::broker::StorageBroker;
 use control_plane::endpoint::{ComputeControlPlane, EndpointTerminateMode};
 use control_plane::endpoint::{
@@ -557,6 +557,9 @@ struct EndpointCreateCmdArgs {
     #[clap(long = "pageserver-id")]
     endpoint_pageserver_id: Option<NodeId>,
 
+    #[clap(long, value_parser = parse_compute_features)]
+    features: Option<ComputeFeatures>,
+
     /// Don't do basebackup, create endpoint directory with only config files.
     #[clap(long, action = clap::ArgAction::Set, default_value_t = false)]
     config_only: bool,
@@ -591,6 +594,13 @@ struct EndpointCreateCmdArgs {
     privileged_role_name: Option<String>,
 }
 
+#[derive(Clone, serde::Deserialize)]
+#[serde(transparent)]
+struct ComputeFeatures(Vec<ComputeFeature>);
+fn parse_compute_features(s: &str) -> anyhow::Result<ComputeFeatures> {
+    serde_json::from_str(s).context("parse compute features arg as JSON array")
+}
+
 /// Start Postgres. If the endpoint doesn't exist yet, it is created.
 #[derive(clap::Args)]
 struct EndpointStartCmdArgs {
@@ -1431,6 +1441,10 @@ async fn handle_endpoint(subcmd: &EndpointCmd, env: &local_env::LocalEnv) -> Res
                 !args.update_catalog,
                 false,
                 args.privileged_role_name.clone(),
+                args.features
+                    .as_ref()
+                    .map(|x| x.0.clone())
+                    .unwrap_or(vec![]),
             )?;
         }
         EndpointCmd::Start(args) => {

control_plane/src/endpoint.rs

@@ -207,6 +207,7 @@ impl ComputeControlPlane {
         skip_pg_catalog_updates: bool,
         drop_subscriptions_before_start: bool,
         privileged_role_name: Option<String>,
+        features: Vec<ComputeFeature>,
     ) -> Result<Arc<Endpoint>> {
         let pg_port = pg_port.unwrap_or_else(|| self.get_port());
         let external_http_port = external_http_port.unwrap_or_else(|| self.get_port() + 1);
@@ -241,7 +242,7 @@ impl ComputeControlPlane {
             drop_subscriptions_before_start,
             grpc,
             reconfigure_concurrency: 1,
-            features: vec![],
+            features: features.clone(),
             cluster: None,
             compute_ctl_config: compute_ctl_config.clone(),
             privileged_role_name: privileged_role_name.clone(),
@@ -263,7 +264,7 @@ impl ComputeControlPlane {
                 skip_pg_catalog_updates,
                 drop_subscriptions_before_start,
                 reconfigure_concurrency: 1,
-                features: vec![],
+                features,
                 cluster: None,
                 compute_ctl_config,
                 privileged_role_name,

control_plane/src/pageserver.rs

@@ -571,6 +571,11 @@ impl PageServerNode {
                 .map(|x| x.parse::<bool>())
                 .transpose()
                 .context("Failed to parse 'basebackup_cache_enabled' as bool")?,
+            standby_horizon_lease_length: settings
+                .remove("standby_horizon_lease_length")
+                .map(humantime::parse_duration)
+                .transpose()
+                .context("Failed to parse 'standby_horizon_lease_length' as duration")?,
         };
         if !settings.is_empty() {
             bail!("Unrecognized tenant settings: {settings:?}")

libs/compute_api/src/spec.rs

@@ -226,6 +226,8 @@ pub enum ComputeFeature {
     /// Enable TLS functionality.
     TlsExperimental,
 
+    StandbyHorizonLeasesExperimental,
+
     /// This is a special feature flag that is used to represent unknown feature flags.
     /// Basically all unknown to enum flags are represented as this one. See unit test
     /// `parse_unknown_features()` for more details.

libs/pageserver_api/src/config.rs

@@ -621,6 +621,10 @@ pub struct TenantConfigToml {
     #[serde(with = "humantime_serde")]
     pub lsn_lease_length_for_ts: Duration,
 
+    /// The length for a standby horizon lease.
+    #[serde(with = "humantime_serde")]
+    pub standby_horizon_lease_length: Duration,
+
     /// Enable auto-offloading of timelines.
     /// (either this flag or the pageserver-global one need to be set)
     pub timeline_offloading: bool,
@@ -950,6 +954,7 @@ impl Default for TenantConfigToml {
             image_creation_preempt_threshold: DEFAULT_IMAGE_CREATION_PREEMPT_THRESHOLD,
             lsn_lease_length: LsnLease::DEFAULT_LENGTH,
             lsn_lease_length_for_ts: LsnLease::DEFAULT_LENGTH_FOR_TS,
+            standby_horizon_lease_length: LsnLease::DEFAULT_STANDBY_HORIZON_LENGTH,
             timeline_offloading: true,
             rel_size_v2_enabled: false,
             gc_compaction_enabled: DEFAULT_GC_COMPACTION_ENABLED,

libs/pageserver_api/src/models.rs

@@ -180,6 +180,9 @@ impl LsnLease {
     /// `get_lsn_by_timestamp` request (1 minutes).
     pub const DEFAULT_LENGTH_FOR_TS: Duration = Duration::from_secs(60);
 
+    /// The default length for a standby horizon lease  (10 minutes).
+    pub const DEFAULT_STANDBY_HORIZON_LENGTH: Duration = Duration::from_secs(10 * 60);
+
     /// Checks whether the lease is expired.
     pub fn is_expired(&self, now: &SystemTime) -> bool {
         now > &self.valid_until
@@ -646,6 +649,8 @@ pub struct TenantConfigPatch {
     pub relsize_snapshot_cache_capacity: FieldPatch<usize>,
     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     pub basebackup_cache_enabled: FieldPatch<bool>,
+    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
+    pub standby_horizon_lease_length: FieldPatch<String>,
 }
 
 /// Like [`crate::config::TenantConfigToml`], but preserves the information
@@ -783,6 +788,10 @@ pub struct TenantConfig {
 
     #[serde(skip_serializing_if = "Option::is_none")]
     pub basebackup_cache_enabled: Option<bool>,
+
+    #[serde(skip_serializing_if = "Option::is_none")]
+    #[serde(with = "humantime_serde")]
+    pub standby_horizon_lease_length: Option<Duration>,
 }
 
 impl TenantConfig {
@@ -830,6 +839,7 @@ impl TenantConfig {
             mut sampling_ratio,
             mut relsize_snapshot_cache_capacity,
             mut basebackup_cache_enabled,
+            mut standby_horizon_lease_length,
         } = self;
 
         patch.checkpoint_distance.apply(&mut checkpoint_distance);
@@ -939,6 +949,10 @@ impl TenantConfig {
         patch
             .basebackup_cache_enabled
             .apply(&mut basebackup_cache_enabled);
+        patch
+            .standby_horizon_lease_length
+            .map(|v| humantime::parse_duration(&v))?
+            .apply(&mut standby_horizon_lease_length);
 
         Ok(Self {
             checkpoint_distance,
@@ -980,6 +994,7 @@ impl TenantConfig {
             sampling_ratio,
             relsize_snapshot_cache_capacity,
             basebackup_cache_enabled,
+            standby_horizon_lease_length,
         })
     }
 
@@ -1094,6 +1109,9 @@ impl TenantConfig {
             basebackup_cache_enabled: self
                 .basebackup_cache_enabled
                 .unwrap_or(global_conf.basebackup_cache_enabled),
+            standby_horizon_lease_length: self
+                .standby_horizon_lease_length
+                .unwrap_or(global_conf.standby_horizon_lease_length),
         }
     }
 }
@@ -1627,6 +1645,9 @@ pub struct TimelineInfo {
     // HADRON: the largest LSN below which all page updates have been included in the image layers.
     #[serde(skip_serializing_if = "Option::is_none")]
     pub image_consistent_lsn: Option<Lsn>,
+
+    #[serde(default)]
+    pub standby_horizon: serde_json::Value,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]

pageserver/page_api/Cargo.toml

@@ -6,6 +6,7 @@ license.workspace = true
 
 [dependencies]
 anyhow.workspace = true
+chrono.workspace = true
 bytes.workspace = true
 futures.workspace = true
 pageserver_api.workspace = true

pageserver/page_api/proto/page_service.proto

@@ -70,6 +70,18 @@ service PageService {
   // Acquires or extends a lease on the given LSN. This guarantees that the Pageserver won't garbage
   // collect the LSN until the lease expires. Must be acquired on all relevant shards.
   rpc LeaseLsn (LeaseLsnRequest) returns (LeaseLsnResponse);
+
+  // Upserts a standby_horizon lease. RO replicas rely on this type of lease.
+  // In slightly more detail: RO replicas always lag to some degree behind the
+  // primary, and request pages at their respective apply LSN. The standby horizon mechanism
+  // ensures that the Pageserver does not garbage-collect old page versions in
+  // the interval between `min(valid standby horizon leases)` and the most recent page version.
+  //
+  // Each RO replica call this method continuously as it applies more WAL.
+  // It identifies its lease through an opaque "lease_id" across these requests.
+  // The response contains the lease expiration time.
+  // Status `FailedPrecondition` is returned if the lease cannot be granted.
+  rpc LeaseStandbyHorizon(LeaseStandbyHorizonRequest) returns (LeaseStandbyHorizonResponse);
 }
 
 // The LSN a request should read at.
@@ -287,3 +299,16 @@ message LeaseLsnResponse {
   // The lease expiration time.
   google.protobuf.Timestamp expires = 1;
 }
+
+// Request for LeaseStandbyHorizon rpc.
+// The lease_id identifies the lease in subsequent requests.
+// The lsn must be monotonic; the request will fail if it is not.
+message LeaseStandbyHorizonRequest {
+  string lease_id  = 1;
+  uint64 lsn = 2;
+}
+
+// Response for the success case of LeaseStandbyHorizon rpc.
+message LeaseStandbyHorizonResponse {
+  google.protobuf.Timestamp expiration = 1;
+}

pageserver/page_api/src/client.rs

@@ -135,6 +135,15 @@ impl Client {
         let resp = self.inner.lease_lsn(req).await?.into_inner();
         Ok(resp.try_into()?)
     }
+
+    pub async fn lease_standby_horizon(
+        &mut self,
+        req: LeaseStandbyHorizonRequest,
+    ) -> tonic::Result<LeaseStandbyHorizonResponse> {
+        let req = proto::LeaseStandbyHorizonRequest::from(req);
+        let resp = self.inner.lease_standby_horizon(req).await?.into_inner();
+        Ok(resp.try_into()?)
+    }
 }
 
 /// Adds authentication metadata to gRPC requests.

pageserver/page_api/src/model.rs

@@ -15,10 +15,13 @@
 //! receivers should expect all sorts of junk from senders. This also allows the sender to use e.g.
 //! stream combinators without dealing with errors, and avoids validating the same message twice.
 
-use std::fmt::Display;
-use std::time::{Duration, SystemTime, UNIX_EPOCH};
+use std::{
+    fmt::Display,
+    time::{Duration, SystemTime, UNIX_EPOCH},
+};
 
 use bytes::Bytes;
+use chrono::Utc;
 use postgres_ffi_types::Oid;
 // TODO: split out Lsn, RelTag, SlruKind and other basic types to a separate crate, to avoid
 // pulling in all of their other crate dependencies when building the client.
@@ -821,3 +824,71 @@ impl From<LeaseLsnResponse> for proto::LeaseLsnResponse {
         }
     }
 }
+
+pub struct LeaseStandbyHorizonRequest {
+    pub lease_id: String,
+    pub lsn: Lsn,
+}
+
+impl TryFrom<proto::LeaseStandbyHorizonRequest> for LeaseStandbyHorizonRequest {
+    type Error = ProtocolError;
+
+    fn try_from(pb: proto::LeaseStandbyHorizonRequest) -> Result<Self, Self::Error> {
+        if pb.lsn == 0 {
+            return Err(ProtocolError::Missing("lsn"));
+        }
+        if pb.lease_id.is_empty() {
+            return Err(ProtocolError::Invalid("lease_id", pb.lease_id));
+        }
+        Ok(Self {
+            lease_id: pb.lease_id,
+            lsn: Lsn(pb.lsn),
+        })
+    }
+}
+
+impl From<LeaseStandbyHorizonRequest> for proto::LeaseStandbyHorizonRequest {
+    fn from(request: LeaseStandbyHorizonRequest) -> Self {
+        Self {
+            lease_id: request.lease_id,
+            lsn: request.lsn.0,
+        }
+    }
+}
+
+/// Lease expiration time. If the lease could not be granted because the LSN has already been
+/// garbage collected, a FailedPrecondition status will be returned instead.
+pub struct LeaseStandbyHorizonResponse {
+    pub expiration: chrono::DateTime<Utc>,
+}
+
+impl TryFrom<proto::LeaseStandbyHorizonResponse> for LeaseStandbyHorizonResponse {
+    type Error = ProtocolError;
+
+    fn try_from(pb: proto::LeaseStandbyHorizonResponse) -> Result<Self, Self::Error> {
+        let expiration = pb.expiration.ok_or(ProtocolError::Missing("expiration"))?;
+        Ok(Self {
+            // TODO: upgrade prost-type to 0.14.1 and use `chrono` feature?
+            expiration: chrono::DateTime::<Utc>::from_timestamp(
+                expiration.seconds,
+                expiration
+                    .nanos
+                    .try_into()
+                    .map_err(|_| ProtocolError::invalid("expiration.nanos", expiration.nanos))?,
+            )
+            .ok_or_else(|| ProtocolError::invalid("expiration", expiration))?,
+        })
+    }
+}
+
+impl From<LeaseStandbyHorizonResponse> for proto::LeaseStandbyHorizonResponse {
+    fn from(response: LeaseStandbyHorizonResponse) -> Self {
+        Self {
+            expiration: Some(prost_types::Timestamp {
+                seconds: response.expiration.timestamp(),
+                nanos: i32::try_from(response.expiration.timestamp_subsec_nanos())
+                    .expect("should fit in i32 max"),
+            }),
+        }
+    }
+}

pageserver/src/config.rs

@@ -656,13 +656,15 @@ impl PageServerConf {
             ..Default::default()
         };
 
-        // Test authors tend to forget about the default 10min initial lease deadline
+        // Test authors tend to forget about the default 10min initial lsn lease deadline
         // when writing tests, which turns their immediate gc requests via mgmt API
         // into no-ops. Override the binary default here, such that there is no initial
-        // lease deadline by default in tests. Tests that care can always override it
+        // lsn lease deadline by default in tests. Tests that care can always override it
         // themselves.
         // Cf https://databricks.atlassian.net/browse/LKB-92?focusedCommentId=6722329
         config_toml.tenant_config.lsn_lease_length = Duration::from_secs(0);
+        // Same argument applies to the initial standby_horizon lease deadline.
+        config_toml.tenant_config.standby_horizon_lease_length = Duration::from_secs(0);
 
         PageServerConf::parse_and_validate(NodeId(0), config_toml, &repo_dir).unwrap()
     }

pageserver/src/http/routes.rs

@@ -486,6 +486,8 @@ async fn build_timeline_info_common(
 
     let (rel_size_migration, rel_size_migrated_at) = timeline.get_rel_size_v2_status();
 
+    let standby_horizon = timeline.standby_horizons.dump();
+
     let info = TimelineInfo {
         tenant_id: timeline.tenant_shard_id,
         timeline_id: timeline.timeline_id,
@@ -524,6 +526,8 @@ async fn build_timeline_info_common(
         walreceiver_status,
         // HADRON
         image_consistent_lsn: None,
+
+        standby_horizon,
     };
     Ok(info)
 }

pageserver/src/metrics.rs

@@ -39,6 +39,7 @@ use crate::tenant::mgr::TenantSlot;
 use crate::tenant::storage_layer::{InMemoryLayer, PersistentLayerDesc};
 use crate::tenant::tasks::BackgroundLoopKind;
 use crate::tenant::throttle::ThrottleResult;
+use crate::tenant::timeline::standby_horizon;
 
 /// Prometheus histogram buckets (in seconds) for operations in the critical
 /// path. In other words, operations that directly affect that latency of user
@@ -723,7 +724,25 @@ static TIMELINE_ARCHIVE_SIZE: Lazy<UIntGaugeVec> = Lazy::new(|| {
 static STANDBY_HORIZON: Lazy<IntGaugeVec> = Lazy::new(|| {
     register_int_gauge_vec!(
         "pageserver_standby_horizon",
-        "Standby apply LSN for which GC is hold off, by timeline.",
+        "Gauge mirroring the legacy standby_horizon propagation mechanism's in-memory value.",
+        &["tenant_id", "shard_id", "timeline_id"]
+    )
+    .expect("failed to define a metric")
+});
+
+static STANDBY_HORIZON_LEASES_COUNT: Lazy<UIntGaugeVec> = Lazy::new(|| {
+    register_uint_gauge_vec!(
+        "pageserver_standby_horizon_leases_count",
+        "Gauge indicating current number of standby horizon leases, per timeline",
+        &["tenant_id", "shard_id", "timeline_id"]
+    )
+    .expect("failed to define a metric")
+});
+
+static STANDBY_HORIZON_LEASES_MIN: Lazy<UIntGaugeVec> = Lazy::new(|| {
+    register_uint_gauge_vec!(
+        "pageserver_standby_horizon_leases_min",
+        "Gauge indicating the minimum of all known standby_horizon lease.",
         &["tenant_id", "shard_id", "timeline_id"]
     )
     .expect("failed to define a metric")
@@ -2327,6 +2346,7 @@ pub(crate) enum ComputeCommandKind {
     Basebackup,
     Fullbackup,
     LeaseLsn,
+    LeaseStandbyHorizon,
 }
 
 pub(crate) struct ComputeCommandCounters {
@@ -3246,7 +3266,7 @@ pub(crate) struct TimelineMetrics {
     pub pitr_history_size: UIntGauge,
     pub archival_size: UIntGauge,
     pub layers_per_read: Histogram,
-    pub standby_horizon_gauge: IntGauge,
+    pub standby_horizon: standby_horizon::Metrics,
     pub resident_physical_size_gauge: UIntGauge,
     pub visible_physical_size_gauge: UIntGauge,
     /// copy of LayeredTimeline.current_logical_size
@@ -3348,9 +3368,17 @@ impl TimelineMetrics {
             .get_metric_with_label_values(&[&tenant_id, &shard_id, &timeline_id])
             .unwrap();
 
-        let standby_horizon_gauge = STANDBY_HORIZON
-            .get_metric_with_label_values(&[&tenant_id, &shard_id, &timeline_id])
-            .unwrap();
+        let standby_horizon = standby_horizon::Metrics {
+            legacy_value: STANDBY_HORIZON
+                .get_metric_with_label_values(&[&tenant_id, &shard_id, &timeline_id])
+                .unwrap(),
+            leases_min: STANDBY_HORIZON_LEASES_MIN
+                .get_metric_with_label_values(&[&tenant_id, &shard_id, &timeline_id])
+                .unwrap(),
+            leases_count: STANDBY_HORIZON_LEASES_COUNT
+                .get_metric_with_label_values(&[&tenant_id, &shard_id, &timeline_id])
+                .unwrap(),
+        };
         let resident_physical_size_gauge = RESIDENT_PHYSICAL_SIZE
             .get_metric_with_label_values(&[&tenant_id, &shard_id, &timeline_id])
             .unwrap();
@@ -3434,7 +3462,7 @@ impl TimelineMetrics {
             pitr_history_size,
             archival_size,
             layers_per_read,
-            standby_horizon_gauge,
+            standby_horizon,
             resident_physical_size_gauge,
             visible_physical_size_gauge,
             current_logical_size_gauge,
@@ -3578,6 +3606,8 @@ impl TimelineMetrics {
         let _ = LAST_RECORD_LSN.remove_label_values(&[tenant_id, shard_id, timeline_id]);
         let _ = DISK_CONSISTENT_LSN.remove_label_values(&[tenant_id, shard_id, timeline_id]);
         let _ = STANDBY_HORIZON.remove_label_values(&[tenant_id, shard_id, timeline_id]);
+        let _ =
+            STANDBY_HORIZON_LEASES_COUNT.remove_label_values(&[tenant_id, shard_id, timeline_id]);
         {
             RESIDENT_PHYSICAL_SIZE_GLOBAL.sub(self.resident_physical_size_get());
             let _ = RESIDENT_PHYSICAL_SIZE.remove_label_values(&[tenant_id, shard_id, timeline_id]);

pageserver/src/page_service.rs

@@ -85,7 +85,7 @@ use crate::tenant::mgr::{
 };
 use crate::tenant::storage_layer::IoConcurrency;
 use crate::tenant::timeline::handle::{Handle, HandleUpgradeError, WeakHandle};
-use crate::tenant::timeline::{self, WaitLsnError, WaitLsnTimeout, WaitLsnWaiter};
+use crate::tenant::timeline::{self, WaitLsnError, WaitLsnTimeout, WaitLsnWaiter, standby_horizon};
 use crate::tenant::{GetTimelineError, PageReconstructError, Timeline};
 use crate::{CancellableTask, PERF_TRACE_TARGET, timed_after_cancellation};
 
@@ -2261,10 +2261,17 @@ impl PageServerHandler {
             ));
         }
 
-        // Clients should only read from recent LSNs on their timeline, or from locations holding an LSN lease.
+        // Reject LSNs for which we may have garbage collected some of the required reconstruct data,
+        // or may do that garbage collection at any point concurrent to this request's execution.
         //
-        // We may have older data available, but we make a best effort to detect this case and return an error,
-        // to distinguish a misbehaving client (asking for old LSN) from a storage issue (data missing at a legitimate LSN).
+        // The only LSNs that are valid to read are:
+        // - While we hold `latest_gc_cutoff_lsn`, the *range* between its value
+        //   and last_record_lsn. (latest_gc_cutoff_lsn is a guard on Timeline::applied_gc_cutoff_lsn).
+        // - An LSN leased via the lsn lease mechanism. Leased LSNs are respected by
+        //   GC but don't prevent applied_gc_cutoff_lsn from advancing.
+        //
+        // NB: Unlike LSN leases, the standby_horizon leases used by RO replicas _do_ prevent the
+        // applied_gc_cutoff_lsn from advancing, so, there's no special treatment for them here.
         if request_lsn < **latest_gc_cutoff_lsn && !timeline.is_gc_blocked_by_lsn_lease_deadline() {
             let gc_info = &timeline.gc_info.read().unwrap();
             if !gc_info.lsn_covered_by_lease(request_lsn) {
@@ -2304,6 +2311,8 @@ impl PageServerHandler {
     where
         IO: AsyncRead + AsyncWrite + Send + Sync + Unpin,
     {
+        debug_assert_current_span_has_tenant_and_timeline_id_no_shard_id();
+
         let timeline = self
             .timeline_handles
             .as_mut()
@@ -2336,7 +2345,7 @@ impl PageServerHandler {
             valid_until_str.as_deref().unwrap_or("<unknown>")
         );
 
-        let bytes = valid_until_str.as_ref().map(|x| x.as_bytes());
+        let bytes: Option<&[u8]> = valid_until_str.as_ref().map(|x| x.as_bytes());
 
         pgb.write_message_noflush(&BeMessage::RowDescription(&[RowDescriptor::text_col(
             b"valid_until",
@@ -2346,6 +2355,48 @@ impl PageServerHandler {
         Ok(())
     }
 
+    #[instrument(skip_all, fields(shard_id, %lsn))]
+    async fn handle_lease_standby_horizon<IO>(
+        &mut self,
+        pgb: &mut PostgresBackend<IO>,
+        tenant_shard_id: TenantShardId,
+        timeline_id: TimelineId,
+        lease_id: String,
+        lsn: Lsn,
+        ctx: &RequestContext,
+    ) -> Result<(), QueryError>
+    where
+        IO: AsyncRead + AsyncWrite + Send + Sync + Unpin,
+    {
+        debug_assert_current_span_has_tenant_and_timeline_id_no_shard_id();
+
+        let timeline = self
+            .timeline_handles
+            .as_mut()
+            .unwrap()
+            .get(
+                tenant_shard_id.tenant_id,
+                timeline_id,
+                ShardSelector::Known(tenant_shard_id.to_index()),
+            )
+            .await?;
+        set_tracing_field_shard_id(&timeline);
+
+        let result = timeline
+            // logs both Ok() and Err() internally, no need to do it here
+            .lease_standby_horizon(lease_id, lsn, ctx)
+            .ok();
+
+        pgb.write_message_noflush(&BeMessage::RowDescription(&[RowDescriptor::text_col(
+            b"expiration",
+        )]))?
+        .write_message_noflush(&BeMessage::DataRow(&[result
+            .map(|standby_horizon::LeaseInfo { valid_until }| valid_until.to_rfc3339().into_bytes())
+            .as_deref()]))?;
+
+        Ok(())
+    }
+
     #[instrument(skip_all, fields(shard_id))]
     async fn handle_get_rel_exists_request(
         timeline: &Timeline,
@@ -2845,6 +2896,14 @@ struct LeaseLsnCmd {
     lsn: Lsn,
 }
 
+#[derive(Debug, Clone, Eq, PartialEq)]
+struct LeaseStandbyHorizonCmd {
+    tenant_shard_id: TenantShardId,
+    timeline_id: TimelineId,
+    lease_id: String,
+    lsn: Lsn,
+}
+
 #[derive(Debug, Clone, Eq, PartialEq)]
 enum PageServiceCmd {
     Set,
@@ -2852,6 +2911,7 @@ enum PageServiceCmd {
     BaseBackup(BaseBackupCmd),
     FullBackup(FullBackupCmd),
     LeaseLsn(LeaseLsnCmd),
+    LeaseStandbyHorizon(LeaseStandbyHorizonCmd),
 }
 
 impl PageStreamCmd {
@@ -3001,6 +3061,31 @@ impl LeaseLsnCmd {
     }
 }
 
+impl LeaseStandbyHorizonCmd {
+    fn parse(query: &str) -> anyhow::Result<Self> {
+        let parameters = query.split_whitespace().collect_vec();
+        if parameters.len() != 4 {
+            bail!(
+                "invalid number of parameters for lease lsn command: {}",
+                query
+            );
+        }
+        let tenant_shard_id = TenantShardId::from_str(parameters[0])
+            .with_context(|| format!("Failed to parse tenant id from {}", parameters[0]))?;
+        let timeline_id = TimelineId::from_str(parameters[1])
+            .with_context(|| format!("Failed to parse timeline id from {}", parameters[1]))?;
+        let lease_id = parameters[2].to_string();
+        let standby_horizon = Lsn::from_str(parameters[3])
+            .with_context(|| format!("Failed to parse lsn from {}", parameters[2]))?;
+        Ok(Self {
+            tenant_shard_id,
+            timeline_id,
+            lease_id,
+            lsn: standby_horizon,
+        })
+    }
+}
+
 impl PageServiceCmd {
     fn parse(query: &str) -> anyhow::Result<Self> {
         let query = query.trim();
@@ -3025,6 +3110,10 @@ impl PageServiceCmd {
                 let cmd2 = cmd2.to_ascii_lowercase();
                 if cmd2 == "lsn" {
                     Ok(Self::LeaseLsn(LeaseLsnCmd::parse(other)?))
+                } else if cmd2 == "standby_horizon" {
+                    Ok(Self::LeaseStandbyHorizon(LeaseStandbyHorizonCmd::parse(
+                        other,
+                    )?))
                 } else {
                     bail!("invalid lease command: {cmd}");
                 }
@@ -3263,7 +3352,7 @@ where
                 lsn,
             }) => {
                 tracing::Span::current()
-                    .record("tenant_id", field::display(tenant_shard_id))
+                    .record("tenant_id", field::display(tenant_shard_id.tenant_id))
                     .record("timeline_id", field::display(timeline_id));
 
                 self.check_permission(Some(tenant_shard_id.tenant_id))?;
@@ -3288,6 +3377,45 @@ where
                     }
                 };
             }
+            PageServiceCmd::LeaseStandbyHorizon(LeaseStandbyHorizonCmd {
+                tenant_shard_id,
+                timeline_id,
+                lease_id,
+                lsn,
+            }) => {
+                tracing::Span::current()
+                    .record("tenant_id", field::display(tenant_shard_id.tenant_id))
+                    .record("timeline_id", field::display(timeline_id));
+
+                self.check_permission(Some(tenant_shard_id.tenant_id))?;
+
+                COMPUTE_COMMANDS_COUNTERS
+                    .for_command(ComputeCommandKind::LeaseStandbyHorizon)
+                    .inc();
+
+                match self
+                    .handle_lease_standby_horizon(
+                        pgb,
+                        tenant_shard_id,
+                        timeline_id,
+                        lease_id,
+                        lsn,
+                        &ctx,
+                    )
+                    .await
+                {
+                    Ok(()) => {
+                        pgb.write_message_noflush(&BeMessage::CommandComplete(b"SELECT 1"))?
+                    }
+                    Err(e) => {
+                        error!("error obtaining standby_horizon lease for {lsn}: {e:?}");
+                        pgb.write_message_noflush(&BeMessage::ErrorResponse(
+                            &e.to_string(),
+                            Some(e.pg_error_code()),
+                        ))?
+                    }
+                };
+            }
         }
 
         Ok(())
@@ -4106,6 +4234,33 @@ impl proto::PageService for GrpcPageServiceHandler {
 
         Ok(tonic::Response::new(expires.into()))
     }
+
+    #[instrument(skip_all, fields(lease_id, lsn))]
+    async fn lease_standby_horizon(
+        &self,
+        req: tonic::Request<proto::LeaseStandbyHorizonRequest>,
+    ) -> Result<tonic::Response<proto::LeaseStandbyHorizonResponse>, tonic::Status> {
+        let timeline = self.get_request_timeline(&req).await?;
+        let ctx = self.ctx.with_scope_timeline(&timeline);
+
+        // Validate and convert the request, and decorate the span.
+        let page_api::LeaseStandbyHorizonRequest { lease_id, lsn } = req.into_inner().try_into()?;
+
+        span_record!(lease_id=%lease_id, lsn=%lsn);
+
+        // Attempt to acquire a lease. Return FailedPrecondition if the lease could not be granted.
+        let standby_horizon::LeaseInfo {
+            valid_until: expiration,
+        } = match timeline.lease_standby_horizon(lease_id, lsn, &ctx) {
+            Ok(expiration) => expiration,
+            // Use Display so the error towards the client is crisp; the function already logged the error with backtrace.
+            Err(err) => return Err(tonic::Status::failed_precondition(format!("{err}"))),
+        };
+
+        Ok(tonic::Response::new(
+            page_api::LeaseStandbyHorizonResponse { expiration }.into(),
+        ))
+    }
 }
 
 /// gRPC middleware layer that handles observability concerns:

pageserver/src/tenant.rs

@@ -179,6 +179,9 @@ pub(super) struct AttachedTenantConf {
     /// The deadline before which we are blocked from GC so that
     /// leases have a chance to be renewed.
     lsn_lease_deadline: Option<tokio::time::Instant>,
+    /// The deadline before which we are blocked from GC so that
+    /// standby horizon leases have a chance to be renewed.
+    standby_horizon_lease_deadline: Option<tokio::time::Instant>,
 }
 
 impl AttachedTenantConf {
@@ -203,10 +206,27 @@ impl AttachedTenantConf {
             None
         };
 
+        // Sets a deadline before which we cannot proceed to GC due to standby horizon lease.
+        //
+        // Similar to LSN leases, standby horizon leases are not persisted to disk. By delaying GC by lease
+        // length, we guarantee that all the standby horizon leases we granted before will have a chance to renew
+        // when we run GC for the first time after restart / transition from AttachedMulti to AttachedSingle.
+        let standby_horizon_lease_deadline = if location.attach_mode == AttachmentMode::Single {
+            Some(
+                tokio::time::Instant::now()
+                    + TenantShard::get_standby_horizon_lease_length_impl(conf, &tenant_conf),
+            )
+        } else {
+            // We don't use `standby_horizon_lease_deadline` to delay GC in AttachedMulti and AttachedStale
+            // because we don't do GC in these modes.
+            None
+        };
+
         Self {
             tenant_conf,
             location,
             lsn_lease_deadline,
+            standby_horizon_lease_deadline,
         }
     }
 
@@ -231,6 +251,12 @@ impl AttachedTenantConf {
             .map(|d| tokio::time::Instant::now() < d)
             .unwrap_or(false)
     }
+
+    fn is_gc_blocked_by_standby_horizon_lease_deadline(&self) -> bool {
+        self.standby_horizon_lease_deadline
+            .map(|d| tokio::time::Instant::now() < d)
+            .unwrap_or(false)
+    }
 }
 struct TimelinePreload {
     timeline_id: TimelineId,
@@ -1580,7 +1606,7 @@ impl TenantShard {
     }
 
     #[instrument(skip_all)]
-    pub(crate) async fn  preload(
+    pub(crate) async fn preload(
         self: &Arc<Self>,
         remote_storage: &GenericRemoteStorage,
         cancel: CancellationToken,
@@ -3097,6 +3123,11 @@ impl TenantShard {
             return Err(GcError::NotActive);
         }
 
+        // Horizon invariants hold at all times, including during startup & initial lease deadline.
+        for tl in self.timelines.lock().unwrap().values().collect_vec() {
+            tl.standby_horizons.validate_invariants(tl);
+        }
+
         {
             let conf = self.tenant_conf.load();
 
@@ -3113,6 +3144,11 @@ impl TenantShard {
                 info!("Skipping GC because lsn lease deadline is not reached");
                 return Ok(GcResult::default());
             }
+
+            if conf.is_gc_blocked_by_standby_horizon_lease_deadline() {
+                info!("Skipping GC because standby horizon lease deadline is not reached");
+                return Ok(GcResult::default());
+            }
         }
 
         let _guard = match self.gc_block.start().await {
@@ -4228,6 +4264,19 @@ impl TenantShard {
             .unwrap_or(conf.default_tenant_conf.lsn_lease_length)
     }
 
+    pub fn get_standby_horizon_lease_length(&self) -> Duration {
+        Self::get_standby_horizon_lease_length_impl(self.conf, &self.tenant_conf.load().tenant_conf)
+    }
+
+    pub fn get_standby_horizon_lease_length_impl(
+        conf: &'static PageServerConf,
+        tenant_conf: &pageserver_api::models::TenantConfig,
+    ) -> Duration {
+        tenant_conf
+            .standby_horizon_lease_length
+            .unwrap_or(conf.default_tenant_conf.standby_horizon_lease_length)
+    }
+
     pub fn get_timeline_offloading_enabled(&self) -> bool {
         if self.conf.timeline_offloading {
             return true;
@@ -4277,6 +4326,7 @@ impl TenantShard {
                     tenant_conf: update(attached_conf.tenant_conf.clone())?,
                     location: attached_conf.location,
                     lsn_lease_deadline: attached_conf.lsn_lease_deadline,
+                    standby_horizon_lease_deadline: attached_conf.standby_horizon_lease_deadline,
                 }))
             })?;
 
@@ -4846,12 +4896,15 @@ impl TenantShard {
 
                 // Cull any expired leases
                 let now = SystemTime::now();
-                target.leases.retain(|_, lease| !lease.is_expired(&now));
+                target.lsn_leases.retain(|_, lease| !lease.is_expired(&now));
+                timeline
+                    .standby_horizons
+                    .cull_leases(tokio::time::Instant::now());
 
                 timeline
                     .metrics
                     .valid_lsn_lease_count_gauge
-                    .set(target.leases.len() as u64);
+                    .set(target.lsn_leases.len() as u64);
 
                 // Look up parent's PITR cutoff to update the child's knowledge of whether it is within parent's PITR
                 if let Some(ancestor_id) = timeline.get_ancestor_timeline_id() {
@@ -6201,6 +6254,7 @@ mod tests {
     use crate::keyspace::KeySpaceAccum;
     use crate::tenant::harness::*;
     use crate::tenant::timeline::CompactFlags;
+    use crate::tenant::timeline::standby_horizon::LeaseInfo;
 
     static TEST_KEY: Lazy<Key> =
         Lazy::new(|| Key::from_slice(&hex!("010000000033333333444444445500000001")));
@@ -9521,7 +9575,7 @@ mod tests {
         // Keeping everything <= Lsn(0x80) b/c leases:
         // 0/10: initdb layer
         // (0/20..=0/70).step_by(0x10): image layers added when creating the timeline.
-        assert_eq!(res.layers_needed_by_leases, 7);
+        assert_eq!(res.layers_needed_by_lsn_leases, 7);
         // Keeping 0/90 b/c it is the latest layer.
         assert_eq!(res.layers_not_updated, 1);
         // Removed 0/80.
@@ -9544,6 +9598,341 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn test_standby_horizons_basics() -> anyhow::Result<()> {
+        let (tenant, ctx) = TenantHarness::create("test_standby_horizons")
+            .await
+            .unwrap()
+            .load()
+            .await;
+
+        // When we call gc_timline below, we have no way to inject a deterministic SystemTime::now()
+        // to the place where it culls leases.
+        // So, pick a lease length longer than this test will run.
+        let lease_length = Duration::from_secs(24 * 60 * 60);
+        tenant
+            .update_tenant_config(|mut conf| {
+                conf.standby_horizon_lease_length = Some(lease_length);
+                Ok(conf)
+            })
+            .unwrap();
+
+        let key = Key::from_hex("010000000033333333444444445500000000").unwrap();
+
+        let end_lsn = Lsn(0x100);
+        let image_layers = (0x10..=0x90)
+            .step_by(0x10)
+            .map(|n| (Lsn(n), vec![(key, test_img(&format!("data key at {n:x}")))]))
+            .collect();
+
+        let timeline = tenant
+            .create_test_timeline_with_layers(
+                TIMELINE_ID,
+                Lsn(0x10),
+                DEFAULT_PG_VERSION,
+                &ctx,
+                Vec::new(), // in-memory layers
+                Vec::new(),
+                image_layers,
+                end_lsn,
+            )
+            .await?;
+
+        // Test lease acquisition from multiple holders, identified by their index in the array.
+        let initial_leases = [0x30, 0x50, 0x70];
+        let mut expirations = Vec::new();
+        initial_leases.iter().enumerate().for_each(|(i, n)| {
+            let lease_id = format!("test_lease_{i}");
+            expirations.push(
+                timeline
+                    .lease_standby_horizon(lease_id, Lsn(*n), &ctx)
+                    .expect("standby horizon lease request should succeed"),
+            );
+        });
+
+        // Test leasing same LSN by different ID yields a fresh lease.
+        // NB: leases are tracked by SystemTime, which is not monotonic, but we want to assert monotonicity of the lease below.
+        // Sleep a second to make flakiness less likely.
+        tokio::time::sleep(Duration::from_secs(1)).await;
+        let LeaseInfo {
+            valid_until: renewed_lease,
+        } = timeline
+            .lease_standby_horizon("renewed_lease_0".to_string(), Lsn(initial_leases[0]), &ctx)
+            .expect("standby horizon lease renewal should succeed");
+        assert!(
+            renewed_lease >= expirations[0].valid_until,
+            "New lease should have expiration time at least as good as original"
+        );
+
+        // whitebox test to assert we're now tracking 4 leases total
+        assert_eq!(timeline.standby_horizons.get_leases().len(), 4);
+        assert_eq!(timeline.standby_horizons.legacy(), None);
+
+        // Also throw in some legacy propagation value at lowest LSN, so we know
+        // legacy propagation is respected.
+        let legacy_lsn = Lsn(0x20);
+        timeline.standby_horizons.register_legacy_update(legacy_lsn);
+        assert_eq!(timeline.standby_horizons.legacy(), Some(legacy_lsn));
+
+        // Force set disk consistent lsn so we can get the cutoff at `end_lsn`.
+        timeline.force_set_disk_consistent_lsn(end_lsn);
+
+        let leases = timeline.standby_horizons.get_leases();
+        let legacy = timeline.standby_horizons.legacy().unwrap();
+
+        //
+        // Do GC with 0 space/time cutoff (horizon=0).
+        //
+        let res = tenant
+            .gc_iteration(
+                Some(TIMELINE_ID),
+                0,
+                Duration::ZERO,
+                &CancellationToken::new(),
+                &ctx,
+            )
+            .await
+            .unwrap();
+        debug!("res: {:?}", res);
+
+        //
+        // Verify standby horizons did hold up GC
+        //
+
+        // 0x20 was the lowest horizon but
+        // TODO cfg(test) forces the new feature right now for coverage.
+        let expected_cutoff = if cfg!(test) || cfg!(feature = "testing") {
+            leases.iter().map(|(lsn, _)| *lsn).min().unwrap()
+        } else {
+            legacy
+        };
+        assert_eq!(*timeline.get_applied_gc_cutoff_lsn(), expected_cutoff);
+
+        // Legacy propagation mechanism gets cleared by gc
+        assert_eq!(timeline.standby_horizons.legacy(), None);
+        // Leases are unaffected.
+        assert_eq!(timeline.standby_horizons.get_leases().len(), 4);
+
+        assert_eq!(
+            res.layers_needed_by_lsn_leases, 0,
+            "we're not using LSN leases in this test"
+        );
+
+        // Should be able to read at the tip
+        timeline
+            .get(key, end_lsn, &ctx)
+            .await
+            .expect("should be able to read data");
+
+        // Should be able to read at any LSN between any standby_horizon and tip
+        let readable_lsns = (legacy.0..=end_lsn.0)
+            .chain(leases.iter().flat_map(|(lsn, _)| (lsn.0..=end_lsn.0)))
+            .dedup()
+            .map(Lsn)
+            .collect_vec();
+        for lsn in readable_lsns {
+            timeline
+                .get(key, lsn, &ctx)
+                .await
+                .expect("should be able to read data");
+        }
+
+        //
+        // Verify culling works
+        // (Again, incorrectly rely on monotonicity of SystemTime here)
+        //
+        timeline
+            .standby_horizons
+            .cull_leases(tokio::time::Instant::now() + lease_length + Duration::from_secs(1));
+
+        assert_eq!(timeline.standby_horizons.get_leases().len(), 0);
+
+        //
+        // Run GC again
+        //
+        let res = tenant
+            .gc_iteration(
+                Some(TIMELINE_ID),
+                0,
+                Duration::ZERO,
+                &CancellationToken::new(),
+                &ctx,
+            )
+            .await
+            .unwrap();
+        debug!("res: {:?}", res);
+
+        // This time, gc should not have been held up.
+        assert_eq!(*timeline.get_applied_gc_cutoff_lsn(), end_lsn);
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_standby_horizon_leases_gc_cutoff_interaction() -> anyhow::Result<()> {
+        let (tenant, ctx) = TenantHarness::create("test_standby_horizons_renewal_below_cutoff")
+            .await
+            .unwrap()
+            .load()
+            .await;
+
+        let timeline = tenant
+            .create_test_timeline(TIMELINE_ID, Lsn(0x10), DEFAULT_PG_VERSION, &ctx)
+            .await?;
+
+        timeline
+            .applied_gc_cutoff_lsn
+            .lock_for_write()
+            .store_and_unlock(Lsn(0x20))
+            .wait()
+            .await;
+
+        // below horizon
+        timeline
+            .lease_standby_horizon("mylease1".to_string(), Lsn(0x10), &ctx)
+            .expect_err("this is below cutoff");
+        timeline.standby_horizons.validate_invariants(&timeline);
+        // at horizon
+        timeline.lease_standby_horizon("mylease2".to_string(), Lsn(0x20), &ctx)?;
+        timeline.standby_horizons.validate_invariants(&timeline);
+        // above horizon
+        timeline.lease_standby_horizon("mylease3".to_string(), Lsn(0x30), &ctx)?;
+
+        timeline.standby_horizons.validate_invariants(&timeline);
+
+        // legacy mechanism had no enforcement to be above gc cutoff in the past
+        timeline.standby_horizons.register_legacy_update(Lsn(0x10));
+        timeline.standby_horizons.validate_invariants(&timeline);
+        timeline.standby_horizons.register_legacy_update(Lsn(0x20));
+        timeline.standby_horizons.validate_invariants(&timeline);
+        timeline.standby_horizons.register_legacy_update(Lsn(0x30));
+        timeline.standby_horizons.validate_invariants(&timeline);
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_standby_horizon_monotonicity() -> anyhow::Result<()> {
+        let (tenant, ctx) = TenantHarness::create("test_standby_horizon_monotonicity")
+            .await
+            .unwrap()
+            .load()
+            .await;
+
+        let timeline = tenant
+            .create_test_timeline(TIMELINE_ID, Lsn(0x10), DEFAULT_PG_VERSION, &ctx)
+            .await?;
+
+        timeline
+            .applied_gc_cutoff_lsn
+            .lock_for_write()
+            .store_and_unlock(Lsn(0x20))
+            .wait()
+            .await;
+
+        // initial
+        timeline.lease_standby_horizon("mylease".to_string(), Lsn(0x20), &ctx)?;
+        timeline.standby_horizons.validate_invariants(&timeline);
+
+        // stay in place
+        timeline.lease_standby_horizon("mylease".to_string(), Lsn(0x20), &ctx)?;
+        timeline.standby_horizons.validate_invariants(&timeline);
+
+        // advance
+        timeline.lease_standby_horizon("mylease".to_string(), Lsn(0x30), &ctx)?;
+        timeline.standby_horizons.validate_invariants(&timeline);
+
+        // move back within gc horizon
+        timeline
+            .lease_standby_horizon("mylease".to_string(), Lsn(0x20), &ctx)
+            .expect_err("should fail");
+        timeline.standby_horizons.validate_invariants(&timeline);
+        let leases = timeline
+            .standby_horizons
+            .get_leases()
+            .into_iter()
+            .map(|(lsn, _)| lsn)
+            .collect_vec();
+        assert_eq!(
+            leases,
+            vec![Lsn(0x30)],
+            "failure should not have changed the lease"
+        );
+
+        // move back below gc horizon also forbidden
+        timeline
+            .lease_standby_horizon("mylease".to_string(), Lsn(0x10), &ctx)
+            .expect_err("should fail");
+        timeline.standby_horizons.validate_invariants(&timeline);
+        let leases = timeline
+            .standby_horizons
+            .get_leases()
+            .into_iter()
+            .map(|(lsn, _)| lsn)
+            .collect_vec();
+        assert_eq!(
+            leases,
+            vec![Lsn(0x30)],
+            "failure should not have changed the lease"
+        );
+
+        // another allowed move forward to ensure no poisoning happened
+        timeline.lease_standby_horizon("mylease".to_string(), Lsn(0x40), &ctx)?;
+        timeline.standby_horizons.validate_invariants(&timeline);
+
+        // legacy can move freely, not subject to monotonicity
+        timeline.standby_horizons.register_legacy_update(Lsn(0x40));
+        timeline.standby_horizons.validate_invariants(&timeline);
+        timeline.standby_horizons.register_legacy_update(Lsn(0x10));
+        timeline.standby_horizons.validate_invariants(&timeline);
+        timeline.standby_horizons.register_legacy_update(Lsn(0x20));
+        timeline.standby_horizons.validate_invariants(&timeline);
+        timeline.standby_horizons.register_legacy_update(Lsn(0x30));
+        timeline.standby_horizons.validate_invariants(&timeline);
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_standby_horizon_emergency_forget_all_leases() -> anyhow::Result<()> {
+        let (tenant, ctx) =
+            TenantHarness::create("test_standby_horizon_emergency_forget_all_leases")
+                .await
+                .unwrap()
+                .load()
+                .await;
+
+        let timeline = tenant
+            .create_test_timeline(TIMELINE_ID, Lsn(0x10), DEFAULT_PG_VERSION, &ctx)
+            .await?;
+
+        //
+        // Empty call works
+        //
+        timeline.standby_horizons.validate_invariants(&timeline);
+        timeline
+            .standby_horizons
+            .emergency_forget_all_leases_immediately();
+        timeline.standby_horizons.validate_invariants(&timeline);
+
+        //
+        // Leases are cleared but legacy remains
+        //
+        let legacy_lsn = Lsn(0x10);
+        timeline.standby_horizons.register_legacy_update(legacy_lsn);
+        timeline.lease_standby_horizon("mylease".to_string(), Lsn(0x20), &ctx)?;
+        timeline.lease_standby_horizon("otherlease".to_string(), Lsn(0x30), &ctx)?;
+        timeline.standby_horizons.validate_invariants(&timeline);
+        timeline
+            .standby_horizons
+            .emergency_forget_all_leases_immediately();
+        timeline.standby_horizons.validate_invariants(&timeline);
+        assert_eq!(timeline.standby_horizons.get_leases().len(), 0);
+        assert_eq!(timeline.standby_horizons.legacy(), Some(legacy_lsn));
+
+        Ok(())
+    }
+
     #[tokio::test]
     async fn test_failed_flush_should_not_update_disk_consistent_lsn() -> anyhow::Result<()> {
         //
@@ -9806,7 +10195,7 @@ mod tests {
                     time: Some(Lsn(0x30)),
                     space: Lsn(0x30),
                 },
-                leases: Default::default(),
+                lsn_leases: Default::default(),
                 within_ancestor_pitr: false,
             };
         }
@@ -10369,7 +10758,7 @@ mod tests {
                     time: Some(Lsn(0x30)),
                     space: Lsn(0x30),
                 },
-                leases: Default::default(),
+                lsn_leases: Default::default(),
                 within_ancestor_pitr: false,
             };
         }
@@ -10634,7 +11023,7 @@ mod tests {
                     time: Some(Lsn(0x30)),
                     space: Lsn(0x30),
                 },
-                leases: Default::default(),
+                lsn_leases: Default::default(),
                 within_ancestor_pitr: false,
             };
         }
@@ -10891,7 +11280,7 @@ mod tests {
                     time: Some(Lsn(0x10)),
                     space: Lsn(0x10),
                 },
-                leases: Default::default(),
+                lsn_leases: Default::default(),
                 within_ancestor_pitr: false,
             };
         }
@@ -10911,7 +11300,7 @@ mod tests {
                     time: Some(Lsn(0x50)),
                     space: Lsn(0x50),
                 },
-                leases: Default::default(),
+                lsn_leases: Default::default(),
                 within_ancestor_pitr: false,
             };
         }
@@ -11637,7 +12026,7 @@ mod tests {
                     time: Some(Lsn(0x30)),
                     space: Lsn(0x30),
                 },
-                leases: Default::default(),
+                lsn_leases: Default::default(),
                 within_ancestor_pitr: false,
             };
         }
@@ -12026,7 +12415,7 @@ mod tests {
                     time: Some(Lsn(0x30)),
                     space: Lsn(0x30),
                 },
-                leases: Default::default(),
+                lsn_leases: Default::default(),
                 within_ancestor_pitr: false,
             };
         }
@@ -12282,7 +12671,7 @@ mod tests {
                     time: Some(Lsn(0x30)),
                     space: Lsn(0x30),
                 },
-                leases: Default::default(),
+                lsn_leases: Default::default(),
                 within_ancestor_pitr: false,
             };
         }
@@ -12612,7 +13001,7 @@ mod tests {
                     time: Some(Lsn(0x30)),
                     space: Lsn(0x30),
                 },
-                leases: Default::default(),
+                lsn_leases: Default::default(),
                 within_ancestor_pitr: false,
             };
         }

pageserver/src/tenant/gc_result.rs

@@ -13,7 +13,7 @@ pub struct GcResult {
     pub layers_needed_by_cutoff: u64,
     pub layers_needed_by_pitr: u64,
     pub layers_needed_by_branches: u64,
-    pub layers_needed_by_leases: u64,
+    pub layers_needed_by_lsn_leases: u64,
     pub layers_not_updated: u64,
     pub layers_removed: u64, // # of layer files removed because they have been made obsolete by newer ondisk files.
 
@@ -43,7 +43,7 @@ impl AddAssign for GcResult {
         self.layers_needed_by_pitr += other.layers_needed_by_pitr;
         self.layers_needed_by_cutoff += other.layers_needed_by_cutoff;
         self.layers_needed_by_branches += other.layers_needed_by_branches;
-        self.layers_needed_by_leases += other.layers_needed_by_leases;
+        self.layers_needed_by_lsn_leases += other.layers_needed_by_lsn_leases;
         self.layers_not_updated += other.layers_not_updated;
         self.layers_removed += other.layers_removed;
 

pageserver/src/tenant/size.rs

@@ -139,8 +139,9 @@ pub struct TimelineInputs {
     /// Cutoff point calculated from the user-supplied 'max_retention_period'
     retention_param_cutoff: Option<Lsn>,
 
-    /// Lease points on the timeline
-    lease_points: Vec<Lsn>,
+    /// LSN lease points on the timeline
+    #[serde(rename = "lease_points")]
+    lsn_lease_points: Vec<Lsn>,
 }
 
 /// Gathers the inputs for the tenant sizing model.
@@ -250,8 +251,8 @@ pub(super) async fn gather_inputs(
 
         let branch_is_invisible = timeline.is_invisible() == Some(true);
 
-        let lease_points = gc_info
-            .leases
+        let lsn_lease_points = gc_info
+            .lsn_leases
             .keys()
             .filter(|&&lsn| lsn > ancestor_lsn)
             .copied()
@@ -274,8 +275,12 @@ pub(super) async fn gather_inputs(
             .collect::<Vec<_>>();
 
         if !branch_is_invisible {
-            // Do not count lease points for invisible branches.
-            lsns.extend(lease_points.iter().map(|&lsn| (lsn, LsnKind::LeasePoint)));
+            // Do not count lsn lease points for invisible branches.
+            lsns.extend(
+                lsn_lease_points
+                    .iter()
+                    .map(|&lsn| (lsn, LsnKind::LeasePoint)),
+            );
         }
 
         drop(gc_info);
@@ -409,7 +414,7 @@ pub(super) async fn gather_inputs(
             latest_gc_cutoff: *timeline.get_applied_gc_cutoff_lsn(),
             next_pitr_cutoff,
             retention_param_cutoff,
-            lease_points,
+            lsn_lease_points,
         });
     }
 

pageserver/src/tenant/timeline.rs

@@ -11,6 +11,7 @@ pub mod layer_manager;
 pub(crate) mod logical_size;
 pub mod offload;
 pub mod span;
+pub(crate) mod standby_horizon;
 pub mod uninit;
 mod walreceiver;
 
@@ -88,7 +89,7 @@ use self::eviction_task::EvictionTaskTimelineState;
 use self::logical_size::LogicalSize;
 use self::walreceiver::{WalReceiver, WalReceiverConf};
 use super::remote_timeline_client::RemoteTimelineClient;
-use super::remote_timeline_client::index::{GcCompactionState, IndexPart, LayerFileMetadata};
+use super::remote_timeline_client::index::{GcCompactionState, IndexPart};
 use super::secondary::heatmap::HeatMapLayer;
 use super::storage_layer::{LayerFringe, LayerVisibilityHint, ReadableLayer};
 use super::tasks::log_compaction_error;
@@ -248,7 +249,7 @@ pub struct Timeline {
     // Atomic would be more appropriate here.
     last_freeze_ts: RwLock<Instant>,
 
-    pub(crate) standby_horizon: AtomicLsn,
+    pub(crate) standby_horizons: standby_horizon::Horizons,
 
     // WAL redo manager. `None` only for broken tenants.
     walredo_mgr: Option<Arc<super::WalRedoManager>>,
@@ -493,7 +494,7 @@ pub(crate) struct GcInfo {
     pub(crate) cutoffs: GcCutoffs,
 
     /// Leases granted to particular LSNs.
-    pub(crate) leases: BTreeMap<Lsn, LsnLease>,
+    pub(crate) lsn_leases: BTreeMap<Lsn, LsnLease>,
 
     /// Whether our branch point is within our ancestor's PITR interval (for cost estimation)
     pub(crate) within_ancestor_pitr: bool,
@@ -541,7 +542,7 @@ impl GcInfo {
         self.remove_child_maybe_offloaded(child_id, MaybeOffloaded::Yes)
     }
     pub(crate) fn lsn_covered_by_lease(&self, lsn: Lsn) -> bool {
-        self.leases.contains_key(&lsn)
+        self.lsn_leases.contains_key(&lsn)
     }
 }
 
@@ -1966,8 +1967,14 @@ impl Timeline {
         _ctx: &RequestContext,
     ) -> anyhow::Result<LsnLease> {
         let lease = {
-            // Normalize the requested LSN to be aligned, and move to the first record
-            // if it points to the beginning of the page (header).
+            // Static (=fixed-lsn) computes request basebackup @ X
+            // but pageserver returns basebackup @ normalize_lsn(X).
+            // Once the postgres process is started, it will therefore
+            // request pages@normalize_lsn(X).
+            // But compute_ctl, which isn't aware that the requested
+            // basebackup lsn != the returned basebackup LSN, will
+            // call this API with X.
+            // So, normalize_lsn(X) here as well.
             let lsn = xlog_utils::normalize_lsn(lsn, WAL_SEGMENT_SIZE);
 
             let mut gc_info = self.gc_info.write().unwrap();
@@ -1975,7 +1982,7 @@ impl Timeline {
 
             let valid_until = SystemTime::now() + length;
 
-            let entry = gc_info.leases.entry(lsn);
+            let entry = gc_info.lsn_leases.entry(lsn);
 
             match entry {
                 Entry::Occupied(mut occupied) => {
@@ -2029,6 +2036,38 @@ impl Timeline {
         Ok(lease)
     }
 
+    /// Logs return value at info level and errors at warn level in Debug::fmt.
+    #[instrument(skip_all, fields(?lease_id, %lsn), ret(level = tracing::Level::INFO, Debug), err(level = tracing::Level::WARN, Debug))]
+    pub(crate) fn lease_standby_horizon(
+        &self,
+        lease_id: String,
+        lsn: Lsn,
+        _ctx: &RequestContext,
+    ) -> anyhow::Result<standby_horizon::LeaseInfo> {
+        if self
+            .feature_resolver
+            .evaluate_boolean("standby-horizon-leases-track-disable")
+            .is_ok()
+        {
+            // Use feature-flagging as an emergency switch in case something is wrong with the data structure (memory consumption, etc.)
+            self.standby_horizons
+                .emergency_forget_all_leases_immediately();
+            return Ok(standby_horizon::LeaseInfo {
+                valid_until: Utc::now() + self.get_standby_horizon_lease_length(),
+            });
+        }
+        let applied_gc_cutoff_lsn_guard = self.get_applied_gc_cutoff_lsn();
+        if lsn < *applied_gc_cutoff_lsn_guard {
+            bail!(
+                "tried to request a standby horizon lease for an lsn below the applied gc cutoff. requested at {} gc cutoff {}",
+                lsn,
+                *applied_gc_cutoff_lsn_guard
+            );
+        }
+        let length = self.get_standby_horizon_lease_length();
+        self.standby_horizons.upsert_lease(lease_id, lsn, length)
+    }
+
     /// Freeze the current open in-memory layer. It will be written to disk on next iteration.
     /// Returns the flush request ID which can be awaited with wait_flush_completion().
     #[instrument(skip(self), fields(tenant_id=%self.tenant_shard_id.tenant_id, shard_id=%self.tenant_shard_id.shard_slug(), timeline_id=%self.timeline_id))]
@@ -2804,6 +2843,14 @@ impl Timeline {
             .unwrap_or(self.conf.default_tenant_conf.lsn_lease_length_for_ts)
     }
 
+    pub(crate) fn get_standby_horizon_lease_length(&self) -> Duration {
+        let tenant_conf = self.tenant_conf.load();
+        tenant_conf
+            .tenant_conf
+            .standby_horizon_lease_length
+            .unwrap_or(self.conf.default_tenant_conf.standby_horizon_lease_length)
+    }
+
     pub(crate) fn is_gc_blocked_by_lsn_lease_deadline(&self) -> bool {
         let tenant_conf = self.tenant_conf.load();
         tenant_conf.is_gc_blocked_by_lsn_lease_deadline()
@@ -3253,8 +3300,6 @@ impl Timeline {
                 ancestor_timeline: ancestor,
                 ancestor_lsn: metadata.ancestor_lsn(),
 
-                metrics,
-
                 query_metrics: crate::metrics::SmgrQueryTimePerTimeline::new(
                     &tenant_shard_id,
                     &timeline_id,
@@ -3319,7 +3364,7 @@ impl Timeline {
                 l0_compaction_trigger: resources.l0_compaction_trigger,
                 gc_lock: tokio::sync::Mutex::default(),
 
-                standby_horizon: AtomicLsn::new(0),
+                standby_horizons: standby_horizon::Horizons::new(metrics.standby_horizon.clone()),
 
                 pagestream_throttle: resources.pagestream_throttle,
 
@@ -3354,6 +3399,8 @@ impl Timeline {
                 feature_resolver: resources.feature_resolver.clone(),
 
                 db_rel_count: ArcSwapOption::from_pointee(None),
+
+                metrics,
             };
 
             result.repartition_threshold =
@@ -4205,12 +4252,6 @@ impl Timeline {
                 let desc: PersistentLayerDesc = hl.name.clone().into();
                 let layer = guard.try_get_from_key(&desc.key())?;
 
-                // Make sure the layer in the old heatmap is the same generation one as in the layer
-                // map otherwise we can in some edge case keep old obsolete layers in the heatmap.
-                if layer.metadata().generation != hl.metadata.generation {
-                    return None;
-                }
-
                 if layer.visibility() == LayerVisibilityHint::Covered {
                     return None;
                 }
@@ -6480,7 +6521,6 @@ pub struct DeltaLayerTestDesc {
     pub lsn_range: Range<Lsn>,
     pub key_range: Range<Key>,
     pub data: Vec<(Key, Lsn, Value)>,
-    pub resident: bool,
 }
 
 #[cfg(test)]
@@ -6498,22 +6538,12 @@ impl DeltaLayerTestDesc {
             lsn_range,
             key_range,
             data,
-            // Default test code creates resident layers.
-            resident: true,
         }
     }
 
     pub fn new_with_inferred_key_range(
         lsn_range: Range<Lsn>,
         data: Vec<(Key, Lsn, Value)>,
-    ) -> Self {
-        Self::new_with_inferred_key_range_and_resident_state(lsn_range, data, true)
-    }
-
-    pub fn new_with_inferred_key_range_and_resident_state(
-        lsn_range: Range<Lsn>,
-        data: Vec<(Key, Lsn, Value)>,
-        resident: bool,
     ) -> Self {
         let key_min = data.iter().map(|(key, _, _)| key).min().unwrap();
         let key_max = data.iter().map(|(key, _, _)| key).max().unwrap();
@@ -6521,7 +6551,6 @@ impl DeltaLayerTestDesc {
             key_range: (*key_min)..(key_max.next()),
             lsn_range,
             data,
-            resident
         }
     }
 
@@ -6800,6 +6829,7 @@ impl Timeline {
     /// Currently, we don't make any attempt at removing unneeded page versions
     /// within a layer file. We can only remove the whole file if it's fully
     /// obsolete.
+    #[instrument(skip_all, fields(standby_horizon_which_min))]
     pub(super) async fn gc(&self) -> Result<GcResult, GcError> {
         // this is most likely the background tasks, but it might be the spawned task from
         // immediate_gc
@@ -6816,7 +6846,7 @@ impl Timeline {
             return Err(GcError::TimelineCancelled);
         }
 
-        let (space_cutoff, time_cutoff, retain_lsns, max_lsn_with_valid_lease) = {
+        let (space_cutoff, time_cutoff, retain_lsns, max_lsn_with_valid_lsn_lease) = {
             let gc_info = self.gc_info.read().unwrap();
 
             let space_cutoff = min(gc_info.cutoffs.space, self.get_disk_consistent_lsn());
@@ -6831,21 +6861,51 @@ impl Timeline {
             //
             // Caveat: `refresh_gc_info` is in charged of updating the lease map.
             // Here, we do not check for stale leases again.
-            let max_lsn_with_valid_lease = gc_info.leases.last_key_value().map(|(lsn, _)| *lsn);
+            let max_lsn_with_valid_lsn_lease =
+                gc_info.lsn_leases.last_key_value().map(|(lsn, _)| *lsn);
 
             (
                 space_cutoff,
                 time_cutoff,
                 retain_lsns,
-                max_lsn_with_valid_lease,
+                max_lsn_with_valid_lsn_lease,
             )
         };
 
         let mut new_gc_cutoff = space_cutoff.min(time_cutoff.unwrap_or_default());
-        let standby_horizon = self.standby_horizon.load();
+
         // Hold GC for the standby, but as a safety guard do it only within some
         // reasonable lag.
-        if standby_horizon != Lsn::INVALID {
+        // TODO: revisit this once we've fully transitioned to leases. 10GiB isn't _that_ much
+        // of permitted lag at high ingest rates. Yet again, unlimited lag is a DoS risk.
+        // Ideally we implement a monitoring function for RO replica lag in a higher level controller,
+        // e.g. the TBD compute manager aka endpoint controller. It would kill the replica if it's
+        // lagging too much or is otherwise unresponsive. The standby horizon leasing could
+        // then be moved into that controller, i.e., inside our trust domain. Replica can still
+        // advance existing leases, but, control over existence (create/destroy lease) would be
+        // with that controller.
+        // When solving this, solve it generically for lsn leases as well.
+        let min_standby_horizon = self.standby_horizons.min_and_clear_legacy();
+        let flag_evaluation_result = self
+            .feature_resolver
+            .evaluate_multivariate("standby-horizon-which-min")
+            .ok();
+        Span::current().record(
+            "standby_horizon_which_min",
+            tracing::field::debug(&flag_evaluation_result),
+        );
+        let min_standby_horizon = if cfg!(test) || cfg!(feature = "testing") {
+            // TODO: parametrize rust test / test suite over the feature flag?
+            // For now, test the new feature. Fix this.
+            min_standby_horizon.leases
+        } else {
+            match flag_evaluation_result.as_deref() {
+                Some("all") => min_standby_horizon.all,
+                Some("leases") => min_standby_horizon.leases,
+                None | Some("legacy") | Some(_) => min_standby_horizon.legacy,
+            }
+        };
+        if let Some(standby_horizon) = min_standby_horizon {
             if let Some(standby_lag) = new_gc_cutoff.checked_sub(standby_horizon) {
                 const MAX_ALLOWED_STANDBY_LAG: u64 = 10u64 << 30; // 10 GB
                 if standby_lag.0 < MAX_ALLOWED_STANDBY_LAG {
@@ -6860,20 +6920,12 @@ impl Timeline {
             }
         }
 
-        // Reset standby horizon to ignore it if it is not updated till next GC.
-        // It is an easy way to unset it when standby disappears without adding
-        // more conf options.
-        self.standby_horizon.store(Lsn::INVALID);
-        self.metrics
-            .standby_horizon_gauge
-            .set(Lsn::INVALID.0 as i64);
-
         let res = self
             .gc_timeline(
                 space_cutoff,
                 time_cutoff,
                 retain_lsns,
-                max_lsn_with_valid_lease,
+                max_lsn_with_valid_lsn_lease,
                 new_gc_cutoff,
             )
             .instrument(
@@ -6892,7 +6944,7 @@ impl Timeline {
         space_cutoff: Lsn,
         time_cutoff: Option<Lsn>, // None if uninitialized
         retain_lsns: Vec<Lsn>,
-        max_lsn_with_valid_lease: Option<Lsn>,
+        max_lsn_with_valid_lsn_lease: Option<Lsn>,
         new_gc_cutoff: Lsn,
     ) -> Result<GcResult, GcError> {
         // FIXME: if there is an ongoing detach_from_ancestor, we should just skip gc
@@ -7008,15 +7060,15 @@ impl Timeline {
                 }
 
                 // 4. Is there a valid lease that requires us to keep this layer?
-                if let Some(lsn) = &max_lsn_with_valid_lease {
+                if let Some(lsn) = &max_lsn_with_valid_lsn_lease {
                     // keep if layer start <= any of the lease
                     if &l.get_lsn_range().start <= lsn {
                         debug!(
-                            "keeping {} because there is a valid lease preventing GC at {}",
+                            "keeping {} because there is a valid LSN lease preventing GC at {}",
                             l.layer_name(),
                             lsn,
                         );
-                        result.layers_needed_by_leases += 1;
+                        result.layers_needed_by_lsn_leases += 1;
                         continue 'outer;
                     }
                 }
@@ -7523,30 +7575,6 @@ impl Timeline {
         check_start_lsn: Option<Lsn>,
         ctx: &RequestContext,
     ) -> anyhow::Result<()> {
-
-        if !deltas.resident {
-            // Don't need to bother creating an on-disk file, we just want the metadata for a non-resident layer.
-            let delta_layer = Layer::for_evicted(
-                self.conf,
-                self,
-                deltas.layer_name(),
-                LayerFileMetadata {
-                    generation: self.generation,
-                    shard: self.get_shard_index(),
-                    file_size: 1024,
-                },
-            );
-            info!("force created non-resident delta layer {}", deltas.layer_name());
-            {
-                let mut guard = self.layers.write(LayerManagerLockHolder::Testing).await;
-                guard
-                    .open_mut()
-                    .unwrap()
-                    .force_insert_optionally_resident_layer(delta_layer);
-            }
-            return Ok(());
-        }
-
         let last_record_lsn = self.get_last_record_lsn();
         deltas
             .data
@@ -8305,148 +8333,6 @@ mod tests {
         ));
     }
 
-    #[tokio::test]
-    async fn test_heatmap_generation_removes_layers_from_old_generation() {
-        use std::time::SystemTime;
-        use utils::generation::Generation;
-        use crate::tenant::remote_timeline_client::index::LayerFileMetadata;
-        use crate::tenant::secondary::heatmap::{HeatMapLayer, HeatMapTimeline as HeatMapTimelineStruct};
-
-        let harness = TenantHarness::create("heatmaheatmap_genereation_removes_layers_from_old_generationp_generation").await.unwrap();
-
-        // Create your existing layer descriptions
-        let covered_delta = DeltaLayerTestDesc::new_with_inferred_key_range(
-            Lsn(0x10)..Lsn(0x20),
-            vec![(
-                Key::from_hex("620000000033333333444444445500000000").unwrap(),
-                Lsn(0x11),
-                Value::Image(test_img("foo")),
-            )],
-        );
-        // This visible layer is non-resident on disk. This is important to reproduce the failure as
-        // a resident file will take priority over the previous heatmap even without the fix for
-        // this issue.
-        let visible_delta = DeltaLayerTestDesc::new_with_inferred_key_range_and_resident_state(
-            Lsn(0x10)..Lsn(0x20),
-            vec![(
-                Key::from_hex("720000000033333333444444445500000000").unwrap(),
-                Lsn(0x11),
-                Value::Image(test_img("foo")),
-            )],
-            false, // Non-resident
-        );
-        let l0_delta = DeltaLayerTestDesc::new(
-            Lsn(0x20)..Lsn(0x30),
-            Key::from_hex("000000000000000000000000000000000000").unwrap()
-                ..Key::from_hex("FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF").unwrap(),
-            vec![(
-                Key::from_hex("720000000033333333444444445500000000").unwrap(),
-                Lsn(0x25),
-                Value::Image(test_img("foo")),
-            )],
-        );
-        let delta_layers = vec![
-            covered_delta.clone(),
-            visible_delta.clone(),
-            l0_delta.clone(),
-        ];
-
-        let image_layer = (
-            Lsn(0x40),
-            vec![(
-                Key::from_hex("620000000033333333444444445500000000").unwrap(),
-                test_img("bar"),
-            )],
-        );
-        let image_layers = vec![image_layer];
-
-        let (tenant, ctx) = harness.load().await;
-
-        // Create timeline with current generation (0xdeadbeef by default)
-        let timeline = tenant
-            .create_test_timeline_with_layers(
-                TimelineId::generate(),
-                Lsn(0x10),
-                PgMajorVersion::PG14,
-                &ctx,
-                Vec::new(), // in-memory layers
-                delta_layers,
-                image_layers,
-                Lsn(0x100),
-            )
-            .await
-            .unwrap();
-
-        // Now create a previous heatmap with the visible_delta layer from an older generation
-        let current_layer_metadata = LayerFileMetadata {
-            generation: timeline.generation,
-            shard: timeline.get_shard_index(),
-            file_size: 1024,
-        };
-        let old_generation = Generation::new(0x12345678); // Older than 0xdeadbeef
-        let old_layer_metadata = LayerFileMetadata {
-            generation: old_generation,
-            shard: timeline.get_shard_index(),
-            file_size: 1024,
-        };
-
-        // Create heatmap layers that reference the same keys but with old generation
-        let prev_heatmap_layers = vec![
-            HeatMapLayer::new(
-                covered_delta.layer_name(),
-                current_layer_metadata.clone(),
-                SystemTime::now(),
-                false, // not cold
-            ),
-            HeatMapLayer::new(
-                visible_delta.layer_name(),
-                // Visible delta layer is from an older generation in heatmap
-                old_layer_metadata.clone(),
-                SystemTime::now(),
-                false, // not cold
-            ),
-            HeatMapLayer::new(
-                l0_delta.layer_name(),
-                current_layer_metadata.clone(),
-                SystemTime::now(),
-                false, // not cold
-            ),
-        ];
-
-        // Create the previous heatmap with old generation layers
-        let prev_heatmap = HeatMapTimelineStruct::new(timeline.timeline_id, prev_heatmap_layers);
-
-        // Set the previous heatmap on the timeline
-        timeline
-            .previous_heatmap
-            .store(Some(Arc::new(PreviousHeatmap::Active {
-                heatmap: prev_heatmap,
-                read_at: std::time::Instant::now(),
-                end_lsn: None,
-            })));
-
-        // Layer visibility is an input to heatmap generation, so refresh it first
-        timeline.update_layer_visibility().await.unwrap();
-
-        // Generate a new heatmap - this should filter out the old generation layers
-        let heatmap = timeline
-            .generate_heatmap()
-            .await
-            .expect("Infallible while timeline is not shut down");
-
-        assert_eq!(heatmap.timeline_id, timeline.timeline_id);
-
-        // Verify that layers exist but they should be the current generation ones,
-        // not the old generation ones from previous_heatmap
-        for layer in heatmap.all_layers() {
-            assert_eq!(
-                layer.metadata.generation,
-                timeline.generation,
-                "Heatmap should only contain current generation layers, not old ones"
-            );
-        }
-    }
-
     #[tokio::test]
     async fn two_layer_eviction_attempts_at_the_same_time() {
         let harness = TenantHarness::create("two_layer_eviction_attempts_at_the_same_time")

pageserver/src/tenant/timeline/compaction.rs

@@ -3353,7 +3353,7 @@ impl Timeline {
                     retain_lsns_below_horizon.push(*lsn);
                 }
             }
-            for lsn in gc_info.leases.keys() {
+            for lsn in gc_info.lsn_leases.keys() {
                 if lsn < &gc_cutoff {
                     retain_lsns_below_horizon.push(*lsn);
                 }

pageserver/src/tenant/timeline/init.rs

@@ -129,16 +129,6 @@ pub(super) fn reconcile(
     // Construct Decisions for layers that are found locally, if they're in remote metadata.  Otherwise
     // construct DismissedLayers to get rid of them.
     for (layer_name, local_metadata) in local_layers {
-        // FIXME: This should probably take generation into account. Currently it's possible to have
-        // an old generation file on disk while a newer one with same name is in index (because
-        // primary just split shard) and we miss that here. We are saved by the check below because
-        // the file size is very likely to be different (and if it isn't then the file contents will
-        // probably be the same anyway in case of shard split), but it's confusing that this logic
-        // doesn't account for name collisions from older generations. Ideally, we should consider a
-        // local file from an older generation than the one in the index to be a different file and
-        // return `DismissedLayer::LocalOnly` if generations don't match. Right now though,
-        // layer_name has the generation part stripped so we'd need to re-parse the generation from
-        // the file name here or back in scan_timeline_dir and add it to LocalLayerFileMetadata.
         let Some(remote_metadata) = index_part.layer_metadata.get(&layer_name) else {
             result.push((layer_name, Err(DismissedLayer::LocalOnly(local_metadata))));
             continue;

pageserver/src/tenant/timeline/layer_manager.rs

@@ -645,13 +645,8 @@ impl OpenLayerManager {
 
     #[cfg(test)]
     pub(crate) fn force_insert_layer(&mut self, layer: ResidentLayer) {
-        self.force_insert_optionally_resident_layer(layer.as_ref().clone());
-    }
-
-    #[cfg(test)]
-    pub(crate) fn force_insert_optionally_resident_layer(&mut self, layer: Layer) {
         let mut updates = self.layer_map.batch_update();
-        Self::insert_historic_layer(layer, &mut updates, &mut self.layer_fmgr);
+        Self::insert_historic_layer(layer.as_ref().clone(), &mut updates, &mut self.layer_fmgr);
         updates.flush()
     }
 

pageserver/src/tenant/timeline/standby_horizon.rs

@@ -0,0 +1,293 @@
+//! The standby horizon functionality is used to ensure that getpage requests from
+//! RO replicas can be served.
+//!
+//! RO replicas always lag to some degree behind the primary, and request pages at
+//! their respective apply LSN. The standby horizon mechanism ensures that the
+//! Pageserver does not garbage-collect old page versions in the interval between
+//! `min(valid standby horizon leases)` and the most recent page version.
+//!
+//! There are currently two ways of how standby horizon is maintained on pageserver:
+//! - legacy: as described in RFC36, replica->safekeeper->broker->pageserver
+//! - leases: TODO
+
+use std::{
+    collections::{HashMap, hash_map},
+    time::Duration,
+};
+
+use chrono::Utc;
+use metrics::{IntGauge, UIntGauge};
+use tokio::time::Instant;
+use tracing::{instrument, warn};
+use utils::lsn::Lsn;
+
+use crate::{assert_u64_eq_usize::UsizeIsU64, tenant::Timeline};
+
+pub struct Horizons {
+    inner: std::sync::Mutex<Inner>,
+}
+struct Inner {
+    legacy: Option<Lsn>,
+    leases_by_id: HashMap<String, Lease>,
+    leases_min: Option<Lsn>,
+    metrics: Metrics,
+}
+
+#[derive(Clone)]
+pub struct Metrics {
+    /// `pageserver_standby_horizon`
+    pub legacy_value: IntGauge,
+    /// `pageserver_standby_horizon_leases_min`
+    pub leases_min: UIntGauge,
+    /// `pageserver_standby_horizon_leases_count`
+    pub leases_count: UIntGauge,
+}
+
+#[derive(Debug)]
+struct Lease {
+    valid_until: tokio::time::Instant,
+    lsn: Lsn,
+}
+
+impl Lease {
+    pub fn try_update(&mut self, update: Lease) -> anyhow::Result<()> {
+        let Lease {
+            valid_until: expiration,
+            lsn,
+        } = update;
+        anyhow::ensure!(self.valid_until <= expiration);
+        anyhow::ensure!(self.lsn <= lsn);
+        *self = update;
+        Ok(())
+    }
+}
+
+#[derive(Debug)]
+pub struct LeaseInfo {
+    pub valid_until: chrono::DateTime<Utc>,
+}
+
+/// Returned by [`Self::min_and_clear_legacy`].
+pub struct Mins {
+    /// Just the legacy mechanism's value.
+    pub legacy: Option<Lsn>,
+    /// Just the leases mechanism's value.
+    pub leases: Option<Lsn>,
+    /// The minimum across legacy and all leases mechanism values.
+    pub all: Option<Lsn>,
+}
+
+impl Horizons {
+    pub fn new(metrics: Metrics) -> Self {
+        let legacy = None;
+        metrics.legacy_value.set(Lsn::INVALID.0 as i64);
+
+        let leases_by_id = HashMap::default();
+        metrics.leases_count.set(0);
+
+        let leases_min = None;
+        metrics.leases_min.set(0);
+
+        Self {
+            inner: std::sync::Mutex::new(Inner {
+                legacy,
+                leases_by_id,
+                leases_min,
+                metrics,
+            }),
+        }
+    }
+
+    /// Register an update via the legacy mechanism.
+    pub fn register_legacy_update(&self, lsn: Lsn) {
+        let mut inner = self.inner.lock().unwrap();
+        inner.legacy = Some(lsn);
+        inner.metrics.legacy_value.set(lsn.0 as i64);
+    }
+
+    /// Get the minimum standby horizon and clear the horizon propagated via the legacy mechanism
+    /// via [`Self::register_legacy_update`].
+    ///
+    /// This method is called from GC to incorporate standby horizons into GC decisions.
+    ///
+    /// The clearing of legacy mechanism state is the way it deals with disappearing replicas.
+    /// The legacy mechanims stops calling [`Self::register_legacy_update`] and so, one GC iteration,
+    /// later, the disappeared replica doesn't affect GC anymore.
+    pub fn min_and_clear_legacy(&self) -> Mins {
+        let mut inner = self.inner.lock().unwrap();
+        let legacy = {
+            inner.metrics.legacy_value.set(Lsn::INVALID.0 as i64);
+            inner.legacy.take()
+        };
+
+        let leases = inner.leases_min;
+
+        let all = std::cmp::min(legacy, inner.leases_min);
+
+        Mins {
+            legacy,
+            leases,
+            all,
+        }
+    }
+
+    /// Renew a lease. Due to their nonpersistent nature we can't tell the difference between
+    /// a renewal and an initial lease, so, let's call this `upsert` as a neutral term.
+    pub fn upsert_lease(
+        &self,
+        id: String,
+        lsn: Lsn,
+        length: Duration,
+    ) -> anyhow::Result<LeaseInfo> {
+        let mut inner = self.inner.lock().unwrap();
+        let now = Instant::now();
+        let valid_until = now + length;
+        let update = Lease { valid_until, lsn };
+        let updated = match inner.leases_by_id.entry(id) {
+            hash_map::Entry::Occupied(mut entry) => {
+                entry.get_mut().try_update(update)?;
+                entry.into_mut()
+            }
+            hash_map::Entry::Vacant(entry) => entry.insert(update),
+        };
+        // Convert the internal expiration time to a wallclock time to return it to the caller.
+        let lease_info = LeaseInfo {
+            valid_until: Utc::now()
+                + updated
+                    .valid_until
+                    // reuse now as the upsert is assumed to be fast
+                    .checked_duration_since(now)
+                    .expect("reuse of now() + guarantee of monotonicity in try_update"),
+        };
+
+        let new_count = inner.leases_by_id.len().into_u64();
+        inner.metrics.leases_count.set(new_count);
+        let leases_min = inner.leases_by_id.values().map(|v| v.lsn).min();
+        inner.leases_min = leases_min;
+        inner.metrics.leases_min.set(leases_min.unwrap_or(Lsn(0)).0);
+
+        Ok(lease_info)
+    }
+
+    pub fn cull_leases(&self, now: Instant) {
+        let mut inner = self.inner.lock().unwrap();
+        let mut min = None;
+        inner.leases_by_id.retain(|_, l| {
+            if l.valid_until > now {
+                let min = min.get_or_insert(l.lsn);
+                *min = std::cmp::min(*min, l.lsn);
+                true
+            } else {
+                false
+            }
+        });
+        inner
+            .metrics
+            .leases_count
+            .set(inner.leases_by_id.len().into_u64());
+        inner.leases_min = min;
+        inner.metrics.leases_min.set(min.unwrap_or(Lsn(0)).0);
+    }
+
+    #[instrument(skip_all)]
+    pub fn emergency_forget_all_leases_immediately(&self) {
+        let mut inner = self.inner.lock().unwrap();
+        for (lease_id, _) in inner.leases_by_id.drain() {
+            warn!("dropping lease early {}", lease_id);
+        }
+        inner
+            .metrics
+            .leases_count
+            .set(inner.leases_by_id.len().into_u64());
+        inner.leases_min = None;
+        inner
+            .metrics
+            .leases_min
+            .set(inner.leases_min.unwrap_or(Lsn(0)).0);
+    }
+
+    pub fn dump(&self) -> serde_json::Value {
+        let inner = self.inner.lock().unwrap();
+        let Inner {
+            legacy,
+            leases_by_id,
+            leases_min,
+            metrics: _,
+        } = &*inner;
+        serde_json::json!({
+            "legacy": format!("{legacy:?}"),
+            "leases_by_id": format!("{leases_by_id:?}"),
+            "leases_min": format!("{leases_min:?}"),
+        })
+    }
+
+    #[instrument(skip_all)]
+    pub fn validate_invariants(&self, timeline: &Timeline) {
+        let mut bug = false;
+        let inner = self.inner.lock().unwrap();
+        let applied_gc_cutoff_lsn = timeline.get_applied_gc_cutoff_lsn();
+
+        // INVARIANT: All leases must be at or above applied gc cutoff.
+        // Violation of this invariant would constitute a bug in gc:
+        // it should
+        for (lease_id, lease) in inner.leases_by_id.iter() {
+            if lease.lsn < *applied_gc_cutoff_lsn {
+                warn!(?lease_id, applied_gc_cutoff_lsn=%*applied_gc_cutoff_lsn, "lease is below the applied gc cutoff");
+                bug = true;
+            }
+        }
+        // The legacy mechanism never had this invariant, so we don't enforce it here.
+        macro_rules! bug_on_neq {
+            ($what:literal, $a:expr, $b:expr,) => {
+                let a = $a;
+                let b = $b;
+                if a != b {
+                    warn!(lhs=?a, rhs=?b, $what);
+                    bug = true;
+                }
+            };
+        }
+
+        // INVARIANT: The lease count metrics is kept in sync
+        bug_on_neq!(
+            "lease count metric",
+            inner.metrics.leases_count.get(),
+            inner.leases_by_id.len().into_u64(),
+        );
+
+        // INVARIANT: The minimum value is the min of all leases
+        bug_on_neq!(
+            "leases_min",
+            inner.leases_min,
+            inner.leases_by_id.values().map(|l| l.lsn).min(),
+        );
+
+        // INVARIANT: The minimum value and the metric is kept in sync
+        bug_on_neq!(
+            "leases_min metric",
+            inner.metrics.leases_min.get(),
+            inner.leases_min.unwrap_or(Lsn(0)).0,
+        );
+
+        // Make tests fail if invariant is violated.
+        if cfg!(test) || cfg!(feature = "testing") {
+            assert!(!bug, "check logs");
+        }
+    }
+
+    #[cfg(test)]
+    pub fn legacy(&self) -> Option<Lsn> {
+        let inner = self.inner.lock().unwrap();
+        inner.legacy
+    }
+
+    #[cfg(test)]
+    pub fn get_leases(&self) -> Vec<(Lsn, tokio::time::Instant)> {
+        let inner = self.inner.lock().unwrap();
+        inner
+            .leases_by_id
+            .values()
+            .map(|Lease { valid_until, lsn }| (*lsn, *valid_until))
+            .collect()
+    }
+}

pageserver/src/tenant/timeline/walreceiver/connection_manager.rs

@@ -754,15 +754,11 @@ impl ConnectionManagerState {
             "safekeeper info update: standby_horizon(cutoff)={}",
             timeline_update.standby_horizon
         );
+        // ignore reports from safekeepers not connected to replicas
         if timeline_update.standby_horizon != 0 {
-            // ignore reports from safekeepers not connected to replicas
             self.timeline
-                .standby_horizon
-                .store(Lsn(timeline_update.standby_horizon));
-            self.timeline
-                .metrics
-                .standby_horizon_gauge
-                .set(timeline_update.standby_horizon as i64);
+                .standby_horizons
+                .register_legacy_update(Lsn(timeline_update.standby_horizon));
         }
 
         let new_safekeeper_id = NodeId(timeline_update.safekeeper_id);

test_runner/fixtures/neon_cli.py

@@ -511,6 +511,7 @@ class NeonLocalCli(AbstractNeonCli):
         allow_multiple=False,
         update_catalog: bool = False,
         privileged_role_name: str | None = None,
+        features: list[str] | None = None,
     ) -> subprocess.CompletedProcess[str]:
         args = [
             "endpoint",
@@ -544,6 +545,8 @@ class NeonLocalCli(AbstractNeonCli):
             args.extend(["--update-catalog"])
         if privileged_role_name is not None:
             args.extend(["--privileged-role-name", privileged_role_name])
+        if features is not None:
+            args.extend(["--features", json.dumps(features)])
 
         res = self.raw_cli(args)
         res.check_returncode()

test_runner/fixtures/neon_fixtures.py

@@ -1321,6 +1321,8 @@ class NeonEnv:
             # themselves.
             # Cf https://databricks.atlassian.net/browse/LKB-92?focusedCommentId=6722329
             tenant_config["lsn_lease_length"] = "0s"
+            # Same argument applies to the initial standby_horizon lease deadline.
+            tenant_config["standby_horizon_lease_length"] = "0s"
 
             if self.pageserver_remote_storage is not None:
                 ps_cfg["remote_storage"] = remote_storage_to_toml_dict(
@@ -4756,6 +4758,7 @@ class Endpoint(PgProtocol, LogUtils):
         allow_multiple: bool = False,
         update_catalog: bool = False,
         privileged_role_name: str | None = None,
+        features: list[str] | None = None,
     ) -> Self:
         """
         Create a new Postgres endpoint.
@@ -4784,6 +4787,7 @@ class Endpoint(PgProtocol, LogUtils):
             allow_multiple=allow_multiple,
             update_catalog=update_catalog,
             privileged_role_name=privileged_role_name,
+            features=features,
         )
         path = Path("endpoints") / self.endpoint_id / "pgdata"
         self.pgdata_dir = self.env.repo_dir / path
@@ -5121,6 +5125,7 @@ class Endpoint(PgProtocol, LogUtils):
         basebackup_request_tries: int | None = None,
         autoprewarm: bool = False,
         offload_lfc_interval_seconds: int | None = None,
+        features: list[str] | None = None,
     ) -> Self:
         """
         Create an endpoint, apply config, and start Postgres.
@@ -5136,6 +5141,7 @@ class Endpoint(PgProtocol, LogUtils):
             lsn=lsn,
             pageserver_id=pageserver_id,
             allow_multiple=allow_multiple,
+            features=features,
         ).start(
             remote_ext_base_url=remote_ext_base_url,
             pageserver_id=pageserver_id,
@@ -5229,6 +5235,7 @@ class EndpointFactory:
         basebackup_request_tries: int | None = None,
         autoprewarm: bool = False,
         offload_lfc_interval_seconds: int | None = None,
+        features: list[str] | None = None,
     ) -> Endpoint:
         ep = Endpoint(
             self.env,
@@ -5252,6 +5259,7 @@ class EndpointFactory:
             basebackup_request_tries=basebackup_request_tries,
             autoprewarm=autoprewarm,
             offload_lfc_interval_seconds=offload_lfc_interval_seconds,
+            features=features,
         )
 
     def create(
@@ -5266,6 +5274,7 @@ class EndpointFactory:
         pageserver_id: int | None = None,
         update_catalog: bool = False,
         privileged_role_name: str | None = None,
+        features: list[str] | None = None,
     ) -> Endpoint:
         ep = Endpoint(
             self.env,
@@ -5290,6 +5299,7 @@ class EndpointFactory:
             pageserver_id=pageserver_id,
             update_catalog=update_catalog,
             privileged_role_name=privileged_role_name,
+            features=features,
         )
 
     def stop_all(self, fail_on_error=True) -> Self:
@@ -5349,6 +5359,7 @@ class EndpointFactory:
         endpoint_id: str | None = None,
         grpc: bool | None = None,
         config_lines: list[str] | None = None,
+        features: list[str] | None = None,
     ) -> Endpoint:
         branch_name = origin.branch_name
         assert origin in self.endpoints
@@ -5362,6 +5373,7 @@ class EndpointFactory:
             grpc=grpc,
             hot_standby=True,
             config_lines=config_lines,
+            features=features,
         )
 
 

test_runner/regress/test_attach_tenant_config.py

@@ -182,6 +182,7 @@ def test_fully_custom_config(positive_env: NeonEnv):
         "image_layer_creation_check_threshold": 1,
         "lsn_lease_length": "1m",
         "lsn_lease_length_for_ts": "5s",
+        "standby_horizon_lease_length": "5s",
         "timeline_offloading": False,
         "rel_size_v2_enabled": True,
         "relsize_snapshot_cache_capacity": 10000,

test_runner/regress/test_hot_standby.py

@@ -136,14 +136,20 @@ def test_hot_standby_gc(neon_env_builder: NeonEnvBuilder, pause_apply: bool):
         # we want to control gc and checkpoint frequency precisely
         "gc_period": "0s",
         "compaction_period": "0s",
+        # this test tests standby_horizon leases feature
+        "standby_horizon_lease_length": "10s",
     }
     env = neon_env_builder.init_start(initial_tenant_conf=tenant_conf)
     timeline_id = env.initial_timeline
     tenant_id = env.initial_tenant
 
+    # enable the compute feature t
+    compute_features = ["standby_horizon_leases_experimental"]
+
     with env.endpoints.create_start(
         branch_name="main",
         endpoint_id="primary",
+        features=compute_features,
     ) as primary:
         with env.endpoints.new_replica_start(
             origin=primary,
@@ -152,6 +158,7 @@ def test_hot_standby_gc(neon_env_builder: NeonEnvBuilder, pause_apply: bool):
             # that this test exercises. With protocol version 1 it
             # fails.
             config_lines=["neon.protocol_version=2"],
+            features=compute_features,
         ) as secondary:
             p_cur = primary.connect().cursor()
             p_cur.execute("CREATE EXTENSION neon_test_utils")