the fix: shutdown method for WalRedoManager, so the Arc<> can outlive the process

This reverts commit 5202d2dc98.

Cargo.lock

@@ -2993,9 +2993,9 @@ checksum = "490cc448043f947bae3cbee9c203358d62dbee0db12107a74be5c30ccfd09771"
 
 [[package]]
 name = "measured"
-version = "0.0.22"
+version = "0.0.21"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3051f3a030d55d680cdef6ca50e80abd1182f8da29f2344a7c9cb575721138f0"
+checksum = "652bc741286361c06de8cb4d89b21a6437f120c508c51713663589eeb9928ac5"
 dependencies = [
  "bytes",
  "crossbeam-utils",
@@ -3011,9 +3011,9 @@ dependencies = [
 
 [[package]]
 name = "measured-derive"
-version = "0.0.22"
+version = "0.0.21"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b9e6777fc80a575f9503d908c8b498782a6c3ee88a06cb416dc3941401e43b94"
+checksum = "6ea497f33e1e856a376c32ad916f69a0bd3c597db1f912a399f842b01a4a685d"
 dependencies = [
  "heck 0.5.0",
  "proc-macro2",
@@ -3023,9 +3023,9 @@ dependencies = [
 
 [[package]]
 name = "measured-process"
-version = "0.0.22"
+version = "0.0.21"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7c4b80445aeb08e832d87bf1830049a924cdc1d6b7ef40b6b9b365bff17bf8ec"
+checksum = "b364ccb66937a814b6b2ad751d1a2f7a9d5a78c761144036825fb36bb0771000"
 dependencies = [
  "libc",
  "measured",

Cargo.toml

@@ -111,8 +111,8 @@ lasso = "0.7"
 leaky-bucket = "1.0.1"
 libc = "0.2"
 md5 = "0.7.0"
-measured = { version = "0.0.22", features=["lasso"] }
-measured-process = { version = "0.0.22" }
+measured = { version = "0.0.21", features=["lasso"] }
+measured-process = { version = "0.0.21" }
 memoffset = "0.8"
 nix = { version = "0.27", features = ["fs", "process", "socket", "signal", "poll"] }
 notify = "6.0.0"

control_plane/storcon_cli/src/main.rs

@@ -1,5 +1,5 @@
 use futures::StreamExt;
-use std::{str::FromStr, time::Duration};
+use std::{collections::HashMap, str::FromStr, time::Duration};
 
 use clap::{Parser, Subcommand};
 use pageserver_api::{
@@ -21,7 +21,7 @@ use utils::id::{NodeId, TenantId};
 
 use pageserver_api::controller_api::{
     NodeConfigureRequest, NodeRegisterRequest, NodeSchedulingPolicy, PlacementPolicy,
-    TenantShardMigrateRequest, TenantShardMigrateResponse,
+    TenantLocateResponse, TenantShardMigrateRequest, TenantShardMigrateResponse,
 };
 
 #[derive(Subcommand, Debug)]
@@ -110,6 +110,12 @@ enum Command {
         #[arg(long)]
         config: String,
     },
+    /// Attempt to balance the locations for a tenant across pageservers.  This is a client-side
+    /// alternative to the storage controller's scheduling optimization behavior.
+    TenantScatter {
+        #[arg(long)]
+        tenant_id: TenantId,
+    },
     /// Print details about a particular tenant, including all its shards' states.
     TenantDescribe {
         #[arg(long)]
@@ -492,6 +498,88 @@ async fn main() -> anyhow::Result<()> {
                 })
                 .await?;
         }
+        Command::TenantScatter { tenant_id } => {
+            // Find the shards
+            let locate_response = storcon_client
+                .dispatch::<(), TenantLocateResponse>(
+                    Method::GET,
+                    format!("control/v1/tenant/{tenant_id}/locate"),
+                    None,
+                )
+                .await?;
+            let shards = locate_response.shards;
+
+            let mut node_to_shards: HashMap<NodeId, Vec<TenantShardId>> = HashMap::new();
+            let shard_count = shards.len();
+            for s in shards {
+                let entry = node_to_shards.entry(s.node_id).or_default();
+                entry.push(s.shard_id);
+            }
+
+            // Load list of available nodes
+            let nodes_resp = storcon_client
+                .dispatch::<(), Vec<NodeDescribeResponse>>(
+                    Method::GET,
+                    "control/v1/node".to_string(),
+                    None,
+                )
+                .await?;
+
+            for node in nodes_resp {
+                if matches!(node.availability, NodeAvailabilityWrapper::Active) {
+                    node_to_shards.entry(node.id).or_default();
+                }
+            }
+
+            let max_shard_per_node = shard_count / node_to_shards.len();
+
+            loop {
+                let mut migrate_shard = None;
+                for shards in node_to_shards.values_mut() {
+                    if shards.len() > max_shard_per_node {
+                        // Pick the emptiest
+                        migrate_shard = Some(shards.pop().unwrap());
+                    }
+                }
+                let Some(migrate_shard) = migrate_shard else {
+                    break;
+                };
+
+                // Pick the emptiest node to migrate to
+                let mut destinations = node_to_shards
+                    .iter()
+                    .map(|(k, v)| (k, v.len()))
+                    .collect::<Vec<_>>();
+                destinations.sort_by_key(|i| i.1);
+                let (destination_node, destination_count) = *destinations.first().unwrap();
+                if destination_count + 1 > max_shard_per_node {
+                    // Even the emptiest destination doesn't have space: we're done
+                    break;
+                }
+                let destination_node = *destination_node;
+
+                node_to_shards
+                    .get_mut(&destination_node)
+                    .unwrap()
+                    .push(migrate_shard);
+
+                println!("Migrate {} -> {} ...", migrate_shard, destination_node);
+
+                storcon_client
+                    .dispatch::<TenantShardMigrateRequest, TenantShardMigrateResponse>(
+                        Method::PUT,
+                        format!("control/v1/tenant/{migrate_shard}/migrate"),
+                        Some(TenantShardMigrateRequest {
+                            tenant_shard_id: migrate_shard,
+                            node_id: destination_node,
+                        }),
+                    )
+                    .await?;
+                println!("Migrate {} -> {} OK", migrate_shard, destination_node);
+            }
+
+            // Spread the shards across the nodes
+        }
         Command::TenantDescribe { tenant_id } => {
             let describe_response = storcon_client
                 .dispatch::<(), TenantDescribeResponse>(

libs/metrics/src/hll.rs

@@ -13,7 +13,11 @@ use std::{
 
 use measured::{
     label::{LabelGroupVisitor, LabelName, LabelValue, LabelVisitor},
-    metric::{gauge::write_gauge, name::MetricNameEncoder, Metric, MetricType, MetricVec},
+    metric::{
+        group::{Encoding, MetricValue},
+        name::MetricNameEncoder,
+        Metric, MetricType, MetricVec,
+    },
     text::TextEncoder,
     LabelGroup,
 };
@@ -178,13 +182,12 @@ impl<W: std::io::Write, const N: usize> measured::metric::MetricEncoding<TextEnc
             .into_iter()
             .enumerate()
             .try_for_each(|(hll_shard, val)| {
-                write_gauge(
-                    enc,
+                enc.write_metric_value(
                     name.by_ref(),
                     labels.by_ref().compose_with(HllShardLabel {
                         hll_shard: hll_shard as i64,
                     }),
-                    val as i64,
+                    MetricValue::Int(val as i64),
                 )
             })
     }

libs/metrics/src/lib.rs

@@ -8,8 +8,8 @@ use measured::{
     label::{LabelGroupSet, LabelGroupVisitor, LabelName, NoLabels},
     metric::{
         counter::CounterState,
-        gauge::{write_gauge, GaugeState},
-        group::Encoding,
+        gauge::GaugeState,
+        group::{Encoding, MetricValue},
         name::{MetricName, MetricNameEncoder},
         MetricEncoding, MetricFamilyEncoding,
     },
@@ -165,6 +165,15 @@ pub struct LibMetrics {
     serve_count: CollectionCounter,
 }
 
+fn write_gauge<Enc: Encoding>(
+    x: i64,
+    labels: impl LabelGroup,
+    name: impl MetricNameEncoder,
+    enc: &mut Enc,
+) -> Result<(), Enc::Err> {
+    enc.write_metric_value(name, labels, MetricValue::Int(x))
+}
+
 #[derive(Default)]
 struct Rusage;
 
@@ -190,12 +199,12 @@ where
             "Bytes written and read from disk, grouped by the operation (read|write)",
         )?;
         GaugeState::write_type(DISK_IO, enc)?;
-        write_gauge(enc, DISK_IO, IoOp::Read, ru.ru_inblock * BYTES_IN_BLOCK)?;
-        write_gauge(enc, DISK_IO, IoOp::Write, ru.ru_oublock * BYTES_IN_BLOCK)?;
+        write_gauge(ru.ru_inblock * BYTES_IN_BLOCK, IoOp::Read, DISK_IO, enc)?;
+        write_gauge(ru.ru_oublock * BYTES_IN_BLOCK, IoOp::Write, DISK_IO, enc)?;
 
         enc.write_help(MAXRSS, "Memory usage (Maximum Resident Set Size)")?;
         GaugeState::write_type(MAXRSS, enc)?;
-        write_gauge(enc, MAXRSS, IoOp::Read, ru.ru_maxrss)?;
+        write_gauge(ru.ru_maxrss, IoOp::Read, MAXRSS, enc)?;
 
         Ok(())
     }
@@ -534,6 +543,15 @@ impl<T: Encoding> Encoding for Inc<T> {
     fn write_help(&mut self, name: impl MetricNameEncoder, help: &str) -> Result<(), Self::Err> {
         self.0.write_help(name, help)
     }
+
+    fn write_metric_value(
+        &mut self,
+        name: impl MetricNameEncoder,
+        labels: impl LabelGroup,
+        value: MetricValue,
+    ) -> Result<(), Self::Err> {
+        self.0.write_metric_value(name, labels, value)
+    }
 }
 
 impl<T: Encoding> MetricEncoding<Inc<T>> for MeasuredCounterPairState
@@ -560,6 +578,15 @@ impl<T: Encoding> Encoding for Dec<T> {
     fn write_help(&mut self, name: impl MetricNameEncoder, help: &str) -> Result<(), Self::Err> {
         self.0.write_help(name, help)
     }
+
+    fn write_metric_value(
+        &mut self,
+        name: impl MetricNameEncoder,
+        labels: impl LabelGroup,
+        value: MetricValue,
+    ) -> Result<(), Self::Err> {
+        self.0.write_metric_value(name, labels, value)
+    }
 }
 
 /// Write the dec counter to the encoder

pageserver/ctl/src/draw_timeline_dir.rs

@@ -83,18 +83,10 @@ fn parse_filename(name: &str) -> (Range<Key>, Range<Lsn>) {
     let keys: Vec<&str> = split[0].split('-').collect();
     let mut lsns: Vec<&str> = split[1].split('-').collect();
 
-    // The current format of the layer file name: 000000067F0000000400000B150100000000-000000067F0000000400000D350100000000__00000000014B7AC8-v1-00000001
-
-    // Handle generation number `-00000001` part
     if lsns.last().expect("should").len() == 8 {
         lsns.pop();
     }
 
-    // Handle version number `-v1` part
-    if lsns.last().expect("should").starts_with('v') {
-        lsns.pop();
-    }
-
     if lsns.len() == 1 {
         lsns.push(lsns[0]);
     }

pageserver/src/bin/pageserver.rs

@@ -394,6 +394,10 @@ fn start_pageserver(
         deletion_workers.spawn_with(BACKGROUND_RUNTIME.handle());
     }
 
+    // Set up global state shared by all walredo processes.
+    let walredo_global_state =
+        BACKGROUND_RUNTIME.block_on(pageserver::walredo::GlobalState::spawn(conf));
+
     // Up to this point no significant I/O has been done: this should have been fast.  Record
     // duration prior to starting I/O intensive phase of startup.
     startup_checkpoint(started_startup_at, "initial", "Starting loading tenants");
@@ -429,6 +433,7 @@ fn start_pageserver(
             broker_client: broker_client.clone(),
             remote_storage: remote_storage.clone(),
             deletion_queue_client,
+            walredo_global_state: Arc::clone(&walredo_global_state),
         },
         order,
         shutdown_pageserver.clone(),
@@ -689,7 +694,13 @@ fn start_pageserver(
             // Right now that tree doesn't reach very far, and `task_mgr` is used instead.
             // The plan is to change that over time.
             shutdown_pageserver.take();
-            pageserver::shutdown_pageserver(&tenant_manager, deletion_queue.clone(), 0).await;
+            pageserver::shutdown_pageserver(
+                &tenant_manager,
+                deletion_queue.clone(),
+                walredo_global_state,
+                0,
+            )
+            .await;
             unreachable!()
         })
     }

pageserver/src/http/openapi_spec.yml

@@ -236,13 +236,6 @@ paths:
             type: string
             format: date-time
           description: A timestamp to get the LSN
-        - name: with_lease
-          in: query
-          required: false
-          schema:
-            type: boolean
-          description: Whether to grant a lease to the corresponding LSN. Default to false.
-
       responses:
         "200":
           description: OK
@@ -1036,10 +1029,6 @@ components:
         kind:
           type: string
           enum: [past, present, future, nodata]
-        valid_until:
-          type: string
-          format: date-time
-          description: The expiration time of the granted lease.
 
     LsnLease:
       type: object

pageserver/src/http/routes.rs

@@ -21,7 +21,6 @@ use pageserver_api::models::IngestAuxFilesRequest;
 use pageserver_api::models::ListAuxFilesRequest;
 use pageserver_api::models::LocationConfig;
 use pageserver_api::models::LocationConfigListResponse;
-use pageserver_api::models::LsnLease;
 use pageserver_api::models::ShardParameters;
 use pageserver_api::models::TenantDetails;
 use pageserver_api::models::TenantLocationConfigResponse;
@@ -729,8 +728,6 @@ async fn get_lsn_by_timestamp_handler(
         .map_err(ApiError::BadRequest)?;
     let timestamp_pg = postgres_ffi::to_pg_timestamp(timestamp);
 
-    let with_lease = parse_query_param(&request, "with_lease")?.unwrap_or(false);
-
     let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);
 
     let timeline =
@@ -739,15 +736,10 @@ async fn get_lsn_by_timestamp_handler(
     let result = timeline
         .find_lsn_for_timestamp(timestamp_pg, &cancel, &ctx)
         .await?;
-
     #[derive(serde::Serialize, Debug)]
     struct Result {
         lsn: Lsn,
         kind: &'static str,
-        #[serde(default)]
-        #[serde(skip_serializing_if = "Option::is_none")]
-        #[serde(flatten)]
-        lease: Option<LsnLease>,
     }
     let (lsn, kind) = match result {
         LsnForTimestamp::Present(lsn) => (lsn, "present"),
@@ -755,28 +747,11 @@ async fn get_lsn_by_timestamp_handler(
         LsnForTimestamp::Past(lsn) => (lsn, "past"),
         LsnForTimestamp::NoData(lsn) => (lsn, "nodata"),
     };
-
-    let lease = if with_lease {
-        timeline
-            .make_lsn_lease(lsn, timeline.get_lsn_lease_length_for_ts(), &ctx)
-            .inspect_err(|_| {
-                warn!("fail to grant a lease to {}", lsn);
-            })
-            .ok()
-    } else {
-        None
-    };
-
-    let result = Result { lsn, kind, lease };
-    let valid_until = result
-        .lease
-        .as_ref()
-        .map(|l| humantime::format_rfc3339_millis(l.valid_until).to_string());
+    let result = Result { lsn, kind };
     tracing::info!(
         lsn=?result.lsn,
         kind=%result.kind,
         timestamp=%timestamp_raw,
-        valid_until=?valid_until,
         "lsn_by_timestamp finished"
     );
     json_response(StatusCode::OK, result)

pageserver/src/lib.rs

@@ -11,6 +11,8 @@ pub mod deletion_queue;
 pub mod disk_usage_eviction_task;
 pub mod http;
 pub mod import_datadir;
+use std::sync::Arc;
+
 pub use pageserver_api::keyspace;
 pub mod aux_file;
 pub mod metrics;
@@ -58,6 +60,7 @@ pub use crate::metrics::preinitialize_metrics;
 pub async fn shutdown_pageserver(
     tenant_manager: &TenantManager,
     mut deletion_queue: DeletionQueue,
+    walredo_global_state: Arc<walredo::GlobalState>,
     exit_code: i32,
 ) {
     use std::time::Duration;
@@ -79,6 +82,18 @@ pub async fn shutdown_pageserver(
     )
     .await;
 
+    // walredo processes are tenant-scoped and should have been shut down after tenant manager shutdown above.
+    //
+    // In practive, we have lingering walredo processes even when pageserver shuts down cleanly, i.e., even when it
+    // does not hit systemd's TimeoutSec timeout (10 seconds in prod).
+    // TODO: understand why the processes aren't gone by the time tenant_manager.shutdown() above returns.
+    timed(
+        walredo_global_state.shutdown(),
+        "wait for all walredo processes to exit",
+        Duration::from_secs(1),
+    )
+    .await;
+
     // Shut down any page service tasks: any in-progress work for particular timelines or tenants
     // should already have been canclled via mgr::shutdown_all_tenants
     timed(

pageserver/src/tenant.rs

@@ -88,6 +88,7 @@ use crate::tenant::remote_timeline_client::MaybeDeletedIndexPart;
 use crate::tenant::remote_timeline_client::INITDB_PATH;
 use crate::tenant::storage_layer::DeltaLayer;
 use crate::tenant::storage_layer::ImageLayer;
+use crate::walredo::Error;
 use crate::InitializationOrder;
 use std::collections::hash_map::Entry;
 use std::collections::BTreeSet;
@@ -165,6 +166,7 @@ pub struct TenantSharedResources {
     pub broker_client: storage_broker::BrokerClientChannel,
     pub remote_storage: GenericRemoteStorage,
     pub deletion_queue_client: DeletionQueueClient,
+    pub walredo_global_state: Arc<crate::walredo::GlobalState>,
 }
 
 /// A [`Tenant`] is really an _attached_ tenant.  The configuration
@@ -323,6 +325,17 @@ impl From<harness::TestRedoManager> for WalRedoManager {
 }
 
 impl WalRedoManager {
+
+    pub(crate) async fn shutdown(&self) {
+        match self {
+            Self::Prod(mgr) => mgr.shutdown().await,
+            #[cfg(test)]
+            Self::Test(mgr) => {
+                // Not applicable to test redo manager
+            }
+        }
+    }
+
     pub(crate) fn maybe_quiesce(&self, idle_timeout: Duration) {
         match self {
             Self::Prod(mgr) => mgr.maybe_quiesce(idle_timeout),
@@ -343,7 +356,7 @@ impl WalRedoManager {
         base_img: Option<(Lsn, bytes::Bytes)>,
         records: Vec<(Lsn, crate::walrecord::NeonWalRecord)>,
         pg_version: u32,
-    ) -> anyhow::Result<bytes::Bytes> {
+    ) -> Result<bytes::Bytes, Error> {
         match self {
             Self::Prod(mgr) => {
                 mgr.request_redo(key, lsn, base_img, records, pg_version)
@@ -649,17 +662,18 @@ impl Tenant {
         mode: SpawnMode,
         ctx: &RequestContext,
     ) -> anyhow::Result<Arc<Tenant>> {
-        let wal_redo_manager = Arc::new(WalRedoManager::from(PostgresRedoManager::new(
-            conf,
-            tenant_shard_id,
-        )));
-
         let TenantSharedResources {
             broker_client,
             remote_storage,
             deletion_queue_client,
+            walredo_global_state,
         } = resources;
 
+        let wal_redo_manager = Arc::new(WalRedoManager::from(PostgresRedoManager::new(
+            walredo_global_state,
+            tenant_shard_id,
+        )));
+
         let attach_mode = attached_conf.location.attach_mode;
         let generation = attached_conf.location.generation;
 
@@ -1877,6 +1891,10 @@ impl Tenant {
         tracing::debug!("Waiting for tasks...");
         task_mgr::shutdown_tasks(None, Some(self.tenant_shard_id), None).await;
 
+        if let Some(walredo_mgr) = self.walredo_mgr.as_ref() {
+            walredo_mgr.shutdown().await;
+        }
+
         // Wait for any in-flight operations to complete
         self.gate.close().await;
 
@@ -3953,7 +3971,7 @@ pub(crate) mod harness {
             base_img: Option<(Lsn, Bytes)>,
             records: Vec<(Lsn, NeonWalRecord)>,
             _pg_version: u32,
-        ) -> anyhow::Result<Bytes> {
+        ) -> Result<Bytes, Error> {
             let records_neon = records.iter().all(|r| apply_neon::can_apply_in_neon(&r.1));
             if records_neon {
                 // For Neon wal records, we can decode without spawning postgres, so do so.
@@ -4007,7 +4025,6 @@ mod tests {
     use storage_layer::PersistentLayerKey;
     use tests::storage_layer::ValuesReconstructState;
     use tests::timeline::{GetVectoredError, ShutdownMode};
-    use timeline::GcInfo;
     use utils::bin_ser::BeSer;
     use utils::id::TenantId;
 
@@ -6685,48 +6702,49 @@ mod tests {
 
         // img layer at 0x10
         let img_layer = (0..10)
-            .map(|id| (get_key(id), Bytes::from(format!("value {id}@0x10"))))
+            .map(|id| (get_key(id), test_img(&format!("value {id}@0x10"))))
             .collect_vec();
 
         let delta1 = vec![
+            // TODO: we should test a real delta record here, which requires us to add a variant of NeonWalRecord for testing purpose.
             (
                 get_key(1),
                 Lsn(0x20),
-                Value::Image(Bytes::from("value 1@0x20")),
+                Value::Image(test_img("value 1@0x20")),
             ),
             (
                 get_key(2),
                 Lsn(0x30),
-                Value::Image(Bytes::from("value 2@0x30")),
+                Value::Image(test_img("value 2@0x30")),
             ),
             (
                 get_key(3),
                 Lsn(0x40),
-                Value::Image(Bytes::from("value 3@0x40")),
+                Value::Image(test_img("value 3@0x40")),
             ),
         ];
         let delta2 = vec![
             (
                 get_key(5),
                 Lsn(0x20),
-                Value::Image(Bytes::from("value 5@0x20")),
+                Value::Image(test_img("value 5@0x20")),
             ),
             (
                 get_key(6),
                 Lsn(0x20),
-                Value::Image(Bytes::from("value 6@0x20")),
+                Value::Image(test_img("value 6@0x20")),
             ),
         ];
         let delta3 = vec![
             (
                 get_key(8),
                 Lsn(0x40),
-                Value::Image(Bytes::from("value 8@0x40")),
+                Value::Image(test_img("value 8@0x40")),
             ),
             (
                 get_key(9),
                 Lsn(0x40),
-                Value::Image(Bytes::from("value 9@0x40")),
+                Value::Image(test_img("value 9@0x40")),
             ),
         ];
 
@@ -6748,42 +6766,9 @@ mod tests {
             guard.cutoffs.horizon = Lsn(0x30);
         }
 
-        let expected_result = [
-            Bytes::from_static(b"value 0@0x10"),
-            Bytes::from_static(b"value 1@0x20"),
-            Bytes::from_static(b"value 2@0x30"),
-            Bytes::from_static(b"value 3@0x40"),
-            Bytes::from_static(b"value 4@0x10"),
-            Bytes::from_static(b"value 5@0x20"),
-            Bytes::from_static(b"value 6@0x20"),
-            Bytes::from_static(b"value 7@0x10"),
-            Bytes::from_static(b"value 8@0x40"),
-            Bytes::from_static(b"value 9@0x40"),
-        ];
-
-        for (idx, expected) in expected_result.iter().enumerate() {
-            assert_eq!(
-                tline
-                    .get(get_key(idx as u32), Lsn(0x50), &ctx)
-                    .await
-                    .unwrap(),
-                expected
-            );
-        }
-
         let cancel = CancellationToken::new();
         tline.compact_with_gc(&cancel, &ctx).await.unwrap();
 
-        for (idx, expected) in expected_result.iter().enumerate() {
-            assert_eq!(
-                tline
-                    .get(get_key(idx as u32), Lsn(0x50), &ctx)
-                    .await
-                    .unwrap(),
-                expected
-            );
-        }
-
         // Check if the image layer at the GC horizon contains exactly what we want
         let image_at_gc_horizon = tline
             .inspect_image_layers(Lsn(0x30), &ctx)
@@ -6794,22 +6779,14 @@ mod tests {
             .collect::<Vec<_>>();
 
         assert_eq!(image_at_gc_horizon.len(), 10);
-        let expected_result = [
-            Bytes::from_static(b"value 0@0x10"),
-            Bytes::from_static(b"value 1@0x20"),
-            Bytes::from_static(b"value 2@0x30"),
-            Bytes::from_static(b"value 3@0x10"),
-            Bytes::from_static(b"value 4@0x10"),
-            Bytes::from_static(b"value 5@0x20"),
-            Bytes::from_static(b"value 6@0x20"),
-            Bytes::from_static(b"value 7@0x10"),
-            Bytes::from_static(b"value 8@0x10"),
-            Bytes::from_static(b"value 9@0x10"),
-        ];
+        let expected_lsn = [0x10, 0x20, 0x30, 0x10, 0x10, 0x20, 0x20, 0x10, 0x10, 0x10];
         for idx in 0..10 {
             assert_eq!(
                 image_at_gc_horizon[idx],
-                (get_key(idx as u32), expected_result[idx].clone())
+                (
+                    get_key(idx as u32),
+                    test_img(&format!("value {idx}@{:#x}", expected_lsn[idx]))
+                )
             );
         }
 
@@ -6842,7 +6819,7 @@ mod tests {
                 },
                 // The delta layer that is cut in the middle
                 PersistentLayerKey {
-                    key_range: get_key(3)..get_key(4),
+                    key_range: Key::MIN..get_key(9),
                     lsn_range: Lsn(0x30)..Lsn(0x41),
                     is_delta: true
                 },
@@ -6927,9 +6904,6 @@ mod tests {
             tline.get(get_key(2), Lsn(0x50), &ctx).await?,
             Bytes::from_static(b"0x10,0x20,0x30")
         );
-
-        // Need to remove the limit of "Neon WAL redo requires base image".
-
         // assert_eq!(tline.get(get_key(3), Lsn(0x50), &ctx).await?, Bytes::new());
         // assert_eq!(tline.get(get_key(4), Lsn(0x50), &ctx).await?, Bytes::new());
 
@@ -7024,164 +6998,4 @@ mod tests {
 
         Ok(())
     }
-
-    #[tokio::test]
-    async fn test_simple_bottom_most_compaction_deltas() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("test_simple_bottom_most_compaction_deltas")?;
-        let (tenant, ctx) = harness.load().await;
-
-        fn get_key(id: u32) -> Key {
-            // using aux key here b/c they are guaranteed to be inside `collect_keyspace`.
-            let mut key = Key::from_hex("620000000033333333444444445500000000").unwrap();
-            key.field6 = id;
-            key
-        }
-
-        // We create one bottom-most image layer, a delta layer D1 crossing the GC horizon, D2 below the horizon, and D3 above the horizon.
-        //
-        //  | D1 |                       | D3 |
-        // -|    |-- gc horizon -----------------
-        //  |    |                | D2 |
-        // --------- img layer ------------------
-        //
-        // What we should expact from this compaction is:
-        //  | Part of D1 |               | D3 |
-        // --------- img layer with D1+D2 at GC horizon------------------
-
-        // img layer at 0x10
-        let img_layer = (0..10)
-            .map(|id| (get_key(id), Bytes::from(format!("value {id}@0x10"))))
-            .collect_vec();
-
-        let delta1 = vec![
-            (
-                get_key(1),
-                Lsn(0x20),
-                Value::WalRecord(NeonWalRecord::wal_append("@0x20")),
-            ),
-            (
-                get_key(2),
-                Lsn(0x30),
-                Value::WalRecord(NeonWalRecord::wal_append("@0x30")),
-            ),
-            (
-                get_key(3),
-                Lsn(0x40),
-                Value::WalRecord(NeonWalRecord::wal_append("@0x40")),
-            ),
-        ];
-        let delta2 = vec![
-            (
-                get_key(5),
-                Lsn(0x20),
-                Value::WalRecord(NeonWalRecord::wal_append("@0x20")),
-            ),
-            (
-                get_key(6),
-                Lsn(0x20),
-                Value::WalRecord(NeonWalRecord::wal_append("@0x20")),
-            ),
-        ];
-        let delta3 = vec![
-            (
-                get_key(8),
-                Lsn(0x40),
-                Value::WalRecord(NeonWalRecord::wal_append("@0x40")),
-            ),
-            (
-                get_key(9),
-                Lsn(0x40),
-                Value::WalRecord(NeonWalRecord::wal_append("@0x40")),
-            ),
-        ];
-
-        let tline = tenant
-            .create_test_timeline_with_layers(
-                TIMELINE_ID,
-                Lsn(0x10),
-                DEFAULT_PG_VERSION,
-                &ctx,
-                vec![delta1, delta2, delta3], // delta layers
-                vec![(Lsn(0x10), img_layer)], // image layers
-                Lsn(0x50),
-            )
-            .await?;
-        {
-            // Update GC info
-            let mut guard = tline.gc_info.write().unwrap();
-            *guard = GcInfo {
-                retain_lsns: vec![],
-                cutoffs: GcCutoffs {
-                    pitr: Lsn(0x30),
-                    horizon: Lsn(0x30),
-                },
-                leases: Default::default(),
-            };
-        }
-
-        let expected_result = [
-            Bytes::from_static(b"value 0@0x10"),
-            Bytes::from_static(b"value 1@0x10@0x20"),
-            Bytes::from_static(b"value 2@0x10@0x30"),
-            Bytes::from_static(b"value 3@0x10@0x40"),
-            Bytes::from_static(b"value 4@0x10"),
-            Bytes::from_static(b"value 5@0x10@0x20"),
-            Bytes::from_static(b"value 6@0x10@0x20"),
-            Bytes::from_static(b"value 7@0x10"),
-            Bytes::from_static(b"value 8@0x10@0x40"),
-            Bytes::from_static(b"value 9@0x10@0x40"),
-        ];
-
-        let expected_result_at_gc_horizon = [
-            Bytes::from_static(b"value 0@0x10"),
-            Bytes::from_static(b"value 1@0x10@0x20"),
-            Bytes::from_static(b"value 2@0x10@0x30"),
-            Bytes::from_static(b"value 3@0x10"),
-            Bytes::from_static(b"value 4@0x10"),
-            Bytes::from_static(b"value 5@0x10@0x20"),
-            Bytes::from_static(b"value 6@0x10@0x20"),
-            Bytes::from_static(b"value 7@0x10"),
-            Bytes::from_static(b"value 8@0x10"),
-            Bytes::from_static(b"value 9@0x10"),
-        ];
-
-        for idx in 0..10 {
-            assert_eq!(
-                tline
-                    .get(get_key(idx as u32), Lsn(0x50), &ctx)
-                    .await
-                    .unwrap(),
-                &expected_result[idx]
-            );
-            assert_eq!(
-                tline
-                    .get(get_key(idx as u32), Lsn(0x30), &ctx)
-                    .await
-                    .unwrap(),
-                &expected_result_at_gc_horizon[idx]
-            );
-        }
-
-        let cancel = CancellationToken::new();
-        tline.compact_with_gc(&cancel, &ctx).await.unwrap();
-
-        for idx in 0..10 {
-            assert_eq!(
-                tline
-                    .get(get_key(idx as u32), Lsn(0x50), &ctx)
-                    .await
-                    .unwrap(),
-                &expected_result[idx]
-            );
-            assert_eq!(
-                tline
-                    .get(get_key(idx as u32), Lsn(0x30), &ctx)
-                    .await
-                    .unwrap(),
-                &expected_result_at_gc_horizon[idx]
-            );
-        }
-
-        Ok(())
-    }
 }

pageserver/src/tenant/timeline.rs

@@ -5235,10 +5235,16 @@ impl Timeline {
                     .map_err(PageReconstructError::WalRedo)?
                     .request_redo(key, request_lsn, data.img, data.records, self.pg_version)
                     .await
-                    .context("reconstruct a page image")
                 {
                     Ok(img) => img,
-                    Err(e) => return Err(PageReconstructError::WalRedo(e)),
+                    Err(e) => {
+                        return Err(match e {
+                            crate::walredo::Error::Cancelled => PageReconstructError::Cancelled,
+                            crate::walredo::Error::Other(e) => {
+                                PageReconstructError::WalRedo(e.context("reconstruct a page image"))
+                            }
+                        })
+                    }
                 };
 
                 Ok(img)

pageserver/src/tenant/timeline/compaction.rs

@@ -965,8 +965,6 @@ impl Timeline {
         _cancel: &CancellationToken,
         ctx: &RequestContext,
     ) -> Result<(), CompactionError> {
-        use std::collections::BTreeSet;
-
         use crate::tenant::storage_layer::ValueReconstructState;
         // Step 0: pick all delta layers + image layers below/intersect with the GC horizon.
         // The layer selection has the following properties:
@@ -988,30 +986,20 @@ impl Timeline {
             (selected_layers, gc_cutoff)
         };
         // Step 1: (In the future) construct a k-merge iterator over all layers. For now, simply collect all keys + LSNs.
-        // Also, collect the layer information to decide when to split the new delta layers.
         let mut all_key_values = Vec::new();
-        let mut delta_split_points = BTreeSet::new();
         for layer in &layer_selection {
             all_key_values.extend(layer.load_key_values(ctx).await?);
-            let desc = layer.layer_desc();
-            if desc.is_delta() {
-                // TODO: is it correct to only record split points for deltas intersecting with the GC horizon? (exclude those below/above the horizon)
-                // so that we can avoid having too many small delta layers.
-                let key_range = desc.get_key_range();
-                delta_split_points.insert(key_range.start);
-                delta_split_points.insert(key_range.end);
-            }
         }
         // Key small to large, LSN low to high, if the same LSN has both image and delta due to the merge of delta layers and
-        // image layers, make image appear before than delta.
+        // image layers, make image appear later than delta.
         struct ValueWrapper<'a>(&'a crate::repository::Value);
         impl Ord for ValueWrapper<'_> {
             fn cmp(&self, other: &Self) -> std::cmp::Ordering {
                 use crate::repository::Value;
                 use std::cmp::Ordering;
                 match (self.0, other.0) {
-                    (Value::Image(_), Value::WalRecord(_)) => Ordering::Less,
-                    (Value::WalRecord(_), Value::Image(_)) => Ordering::Greater,
+                    (Value::Image(_), Value::WalRecord(_)) => Ordering::Greater,
+                    (Value::WalRecord(_), Value::Image(_)) => Ordering::Less,
                     _ => Ordering::Equal,
                 }
             }
@@ -1030,6 +1018,13 @@ impl Timeline {
         all_key_values.sort_by(|(k1, l1, v1), (k2, l2, v2)| {
             (k1, l1, ValueWrapper(v1)).cmp(&(k2, l2, ValueWrapper(v2)))
         });
+        let max_lsn = all_key_values
+            .iter()
+            .map(|(_, lsn, _)| lsn)
+            .max()
+            .copied()
+            .unwrap()
+            + 1;
         // Step 2: Produce images+deltas. TODO: ensure newly-produced delta does not overlap with other deltas.
         // Data of the same key.
         let mut accumulated_values = Vec::new();
@@ -1048,19 +1043,7 @@ impl Timeline {
             // We have a list of deltas/images. We want to create image layers while collect garbages.
             for (key, lsn, val) in accumulated_values.iter().rev() {
                 if *lsn > horizon {
-                    if let Some((_, prev_lsn, _)) = keys_above_horizon.last_mut() {
-                        if *prev_lsn == *lsn {
-                            // The case that we have an LSN with both data from the delta layer and the image layer. As
-                            // `ValueWrapper` ensures that an image is ordered before a delta at the same LSN, we simply
-                            // drop this delta and keep the image.
-                            //
-                            // For example, we have delta layer key1@0x10, key1@0x20, and image layer key1@0x10, we will
-                            // keep the image for key1@0x10 and the delta for key1@0x20. key1@0x10 delta will be simply
-                            // dropped.
-                            continue;
-                        }
-                    }
-                    keys_above_horizon.push((*key, *lsn, val.clone()));
+                    keys_above_horizon.push((*key, *lsn, val.clone())); // TODO: ensure one LSN corresponds to either delta or image instead of both
                 } else if *lsn <= horizon {
                     match val {
                         crate::repository::Value::Image(image) => {
@@ -1085,59 +1068,15 @@ impl Timeline {
             Ok((keys_above_horizon, img))
         }
 
-        async fn flush_deltas(
-            deltas: &mut Vec<(Key, Lsn, crate::repository::Value)>,
-            last_key: Key,
-            delta_split_points: &[Key],
-            current_delta_split_point: &mut usize,
-            tline: &Arc<Timeline>,
-            gc_cutoff: Lsn,
-            ctx: &RequestContext,
-        ) -> anyhow::Result<Option<ResidentLayer>> {
-            // Check if we need to split the delta layer. We split at the original delta layer boundary to avoid
-            // overlapping layers.
-            //
-            // If we have a structure like this:
-            //
-            // | Delta 1 |         | Delta 4 |
-            // |---------| Delta 2 |---------|
-            // | Delta 3 |         | Delta 5 |
-            //
-            // And we choose to compact delta 2+3+5. We will get an overlapping delta layer with delta 1+4.
-            // A simple solution here is to split the delta layers using the original boundary, while this
-            // might produce a lot of small layers. This should be improved and fixed in the future.
-            let mut need_split = false;
-            while *current_delta_split_point < delta_split_points.len()
-                && last_key >= delta_split_points[*current_delta_split_point]
-            {
-                *current_delta_split_point += 1;
-                need_split = true;
-            }
-            if !need_split {
-                return Ok(None);
-            }
-            let deltas = std::mem::take(deltas);
-            if deltas.is_empty() {
-                return Ok(None);
-            }
-            let end_lsn = deltas.iter().map(|(_, lsn, _)| lsn).max().copied().unwrap() + 1;
-            let mut delta_layer_writer = DeltaLayerWriter::new(
-                tline.conf,
-                tline.timeline_id,
-                tline.tenant_shard_id,
-                deltas.first().unwrap().0,
-                gc_cutoff..end_lsn,
-                ctx,
-            )
-            .await?;
-            let key_end = deltas.last().unwrap().0.next();
-            for (key, lsn, val) in deltas {
-                delta_layer_writer.put_value(key, lsn, val, ctx).await?;
-            }
-            let delta_layer = delta_layer_writer.finish(key_end, tline, ctx).await?;
-            Ok(Some(delta_layer))
-        }
-
+        let mut delta_layer_writer = DeltaLayerWriter::new(
+            self.conf,
+            self.timeline_id,
+            self.tenant_shard_id,
+            all_key_values.first().unwrap().0,
+            gc_cutoff..max_lsn, // TODO: off by one?
+            ctx,
+        )
+        .await?;
         let mut image_layer_writer = ImageLayerWriter::new(
             self.conf,
             self.timeline_id,
@@ -1148,10 +1087,6 @@ impl Timeline {
         )
         .await?;
 
-        let mut delta_values = Vec::new();
-        let delta_split_points = delta_split_points.into_iter().collect_vec();
-        let mut current_delta_split_point = 0;
-        let mut delta_layers = Vec::new();
         for item @ (key, _, _) in &all_key_values {
             if &last_key == key {
                 accumulated_values.push(item);
@@ -1159,54 +1094,33 @@ impl Timeline {
                 let (deltas, image) =
                     flush_accumulated_states(self, last_key, &accumulated_values, gc_cutoff)
                         .await?;
-                // Put the image into the image layer. Currently we have a single big layer for the compaction.
                 image_layer_writer.put_image(last_key, image, ctx).await?;
-                delta_values.extend(deltas);
-                delta_layers.extend(
-                    flush_deltas(
-                        &mut delta_values,
-                        last_key,
-                        &delta_split_points,
-                        &mut current_delta_split_point,
-                        self,
-                        gc_cutoff,
-                        ctx,
-                    )
-                    .await?,
-                );
+                for (key, lsn, val) in deltas {
+                    delta_layer_writer.put_value(key, lsn, val, ctx).await?;
+                }
                 accumulated_values.clear();
                 accumulated_values.push(item);
                 last_key = *key;
             }
         }
-
-        // TODO: move this part to the loop body
         let (deltas, image) =
             flush_accumulated_states(self, last_key, &accumulated_values, gc_cutoff).await?;
-        // Put the image into the image layer. Currently we have a single big layer for the compaction.
         image_layer_writer.put_image(last_key, image, ctx).await?;
-        delta_values.extend(deltas);
-        delta_layers.extend(
-            flush_deltas(
-                &mut delta_values,
-                last_key,
-                &delta_split_points,
-                &mut current_delta_split_point,
-                self,
-                gc_cutoff,
-                ctx,
-            )
-            .await?,
-        );
-
+        for (key, lsn, val) in deltas {
+            delta_layer_writer.put_value(key, lsn, val, ctx).await?;
+        }
+        accumulated_values.clear();
+        // TODO: split layers
+        let delta_layer = delta_layer_writer.finish(last_key, self, ctx).await?;
         let image_layer = image_layer_writer.finish(self, ctx).await?;
-        let mut compact_to = Vec::new();
-        compact_to.extend(delta_layers);
-        compact_to.push(image_layer);
         // Step 3: Place back to the layer map.
         {
             let mut guard = self.layers.write().await;
-            guard.finish_gc_compaction(&layer_selection, &compact_to, &self.metrics)
+            guard.finish_gc_compaction(
+                &layer_selection,
+                &[delta_layer.clone(), image_layer.clone()],
+                &self.metrics,
+            )
         };
         Ok(())
     }

pageserver/src/walredo.rs

@@ -40,8 +40,30 @@ use std::time::Duration;
 use std::time::Instant;
 use tracing::*;
 use utils::lsn::Lsn;
+use utils::sync::gate::Gate;
 use utils::sync::heavier_once_cell;
 
+pub struct GlobalState {
+    conf: &'static PageServerConf,
+    pub(self) spawn_gate: Gate,
+}
+
+impl GlobalState {
+    pub async fn spawn(conf: &'static PageServerConf) -> Arc<GlobalState> {
+        let state = Arc::new(GlobalState {
+            conf,
+            spawn_gate: Gate::default(),
+        });
+        state
+    }
+    pub(crate) async fn shutdown(self: &Arc<Self>) {
+        self.spawn_gate.close().await
+        // The destructor of WalRedoProcess SIGKILLs and `wait()`s for the process
+        // The gate guard is stored in WalRedoProcess.
+        // So, we arrive here once all WalRedoProcess structs are gone.
+    }
+}
+
 ///
 /// This is the real implementation that uses a Postgres process to
 /// perform WAL replay. Only one thread can use the process at a time,
@@ -50,8 +72,8 @@ use utils::sync::heavier_once_cell;
 /// records.
 ///
 pub struct PostgresRedoManager {
+    global_state: Arc<GlobalState>,
     tenant_shard_id: TenantShardId,
-    conf: &'static PageServerConf,
     last_redo_at: std::sync::Mutex<Option<Instant>>,
     /// The current [`process::WalRedoProcess`] that is used by new redo requests.
     /// We use [`heavier_once_cell`] for coalescing the spawning, but the redo
@@ -65,9 +87,30 @@ pub struct PostgresRedoManager {
     /// still be using the old redo process. But, those other tasks will most likely
     /// encounter an error as well, and errors are an unexpected condition anyway.
     /// So, probably we could get rid of the `Arc` in the future.
-    redo_process: heavier_once_cell::OnceCell<Arc<process::WalRedoProcess>>,
+    redo_process: heavier_once_cell::OnceCell<RedoProcessState>,
+    launch_process_gate: Gate,
 }
 
+enum RedoProcessState {
+    Launched(Arc<process::WalRedoProcess>),
+    ManagerShutDown,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+    #[error("cancelled")]
+    Cancelled,
+    #[error(transparent)]
+    Other(#[from] anyhow::Error),
+}
+
+macro_rules! bail {
+    ($($arg:tt)*) => {
+        return Err($crate::walredo::Error::Other(::anyhow::anyhow!($($arg)*)));
+    }
+}
+pub(self) use bail;
+
 ///
 /// Public interface of WAL redo manager
 ///
@@ -88,9 +131,9 @@ impl PostgresRedoManager {
         base_img: Option<(Lsn, Bytes)>,
         records: Vec<(Lsn, NeonWalRecord)>,
         pg_version: u32,
-    ) -> anyhow::Result<Bytes> {
+    ) -> Result<Bytes, Error> {
         if records.is_empty() {
-            anyhow::bail!("invalid WAL redo request with no records");
+            bail!("invalid WAL redo request with no records");
         }
 
         let base_img_lsn = base_img.as_ref().map(|p| p.0).unwrap_or(Lsn::INVALID);
@@ -110,7 +153,7 @@ impl PostgresRedoManager {
                         img,
                         base_img_lsn,
                         &records[batch_start..i],
-                        self.conf.wal_redo_timeout,
+                        self.global_state.conf.wal_redo_timeout,
                         pg_version,
                     )
                     .await
@@ -131,7 +174,7 @@ impl PostgresRedoManager {
                 img,
                 base_img_lsn,
                 &records[batch_start..],
-                self.conf.wal_redo_timeout,
+                self.global_state.conf.wal_redo_timeout,
                 pg_version,
             )
             .await
@@ -148,10 +191,10 @@ impl PostgresRedoManager {
                     chrono::Utc::now().checked_sub_signed(chrono::Duration::from_std(age).ok()?)
                 })
             },
-            process: self
-                .redo_process
-                .get()
-                .map(|p| WalRedoManagerProcessStatus { pid: p.id() }),
+            process: self.redo_process.get().and_then(|p| match &*p {
+                RedoProcessState::Launched(p) => Some(WalRedoManagerProcessStatus { pid: p.id() }),
+                RedoProcessState::ManagerShutDown => None,
+            }),
         }
     }
 }
@@ -161,18 +204,36 @@ impl PostgresRedoManager {
     /// Create a new PostgresRedoManager.
     ///
     pub fn new(
-        conf: &'static PageServerConf,
+        global_state: Arc<GlobalState>,
         tenant_shard_id: TenantShardId,
     ) -> PostgresRedoManager {
         // The actual process is launched lazily, on first request.
         PostgresRedoManager {
+            global_state,
             tenant_shard_id,
-            conf,
             last_redo_at: std::sync::Mutex::default(),
             redo_process: heavier_once_cell::OnceCell::default(),
+            launch_process_gate: Gate::default(),
         }
     }
 
+    pub async fn shutdown(&self) {
+        // prevent new launches
+        let permit = match self.redo_process.get_or_init_detached().await {
+            Ok(guard) => {
+                let (proc, permit) = guard.take_and_deinit();
+                drop(proc); // this just drops the Arc, its refcount may not be zero yet
+                permit
+            }
+            Err(permit) => permit,
+        };
+        self.redo_process
+            .set(RedoProcessState::ManagerShutDown, permit);
+
+        // wait for all WalRedoProcess objects to get dropped
+        self.launch_process_gate.close().await;
+    }
+
     /// This type doesn't have its own background task to check for idleness: we
     /// rely on our owner calling this function periodically in its own housekeeping
     /// loops.
@@ -203,7 +264,7 @@ impl PostgresRedoManager {
         records: &[(Lsn, NeonWalRecord)],
         wal_redo_timeout: Duration,
         pg_version: u32,
-    ) -> anyhow::Result<Bytes> {
+    ) -> Result<Bytes, Error> {
         *(self.last_redo_at.lock().unwrap()) = Some(Instant::now());
 
         let (rel, blknum) = key.to_rel_block().context("invalid record")?;
@@ -212,17 +273,27 @@ impl PostgresRedoManager {
         loop {
             let proc: Arc<process::WalRedoProcess> =
                 match self.redo_process.get_or_init_detached().await {
-                    Ok(guard) => Arc::clone(&guard),
+                    Ok(guard) => match &*guard {
+                        RedoProcessState::Launched(proc) => Arc::clone(proc),
+                        RedoProcessState::ManagerShutDown => {
+                            return Err(Error::Cancelled);
+                        }
+                    },
                     Err(permit) => {
                         // don't hold poison_guard, the launch code can bail
                         let start = Instant::now();
                         let proc = Arc::new(
                             process::WalRedoProcess::launch(
-                                self.conf,
+                                &self.global_state,
                                 self.tenant_shard_id,
                                 pg_version,
                             )
-                            .context("launch walredo process")?,
+                            .map_err(|e| match e {
+                                process::LaunchError::Cancelled => Error::Cancelled,
+                                process::LaunchError::Other(e) => {
+                                    Error::Other(e.context("launch walredo process"))
+                                }
+                            })?,
                         );
                         let duration = start.elapsed();
                         WAL_REDO_PROCESS_LAUNCH_DURATION_HISTOGRAM.observe(duration.as_secs_f64());
@@ -231,7 +302,8 @@ impl PostgresRedoManager {
                             pid = proc.id(),
                             "launched walredo process"
                         );
-                        self.redo_process.set(Arc::clone(&proc), permit);
+                        self.redo_process
+                            .set(RedoProcessState::Launched(Arc::clone(&proc)), permit);
                         proc
                     }
                 };
@@ -242,7 +314,14 @@ impl PostgresRedoManager {
             let result = proc
                 .apply_wal_records(rel, blknum, &base_img, records, wal_redo_timeout)
                 .await
-                .context("apply_wal_records");
+                .map_err(|e| match e {
+                    Error::Cancelled => Error::Cancelled,
+                    Error::Other(e) => Error::Other(e.context("apply_wal_records")),
+                });
+            if matches!(result, Err(Error::Cancelled)) {
+                // bail asap and also avoid log noise due to the error reporting below
+                return Err(Error::Cancelled);
+            }
 
             let duration = started_at.elapsed();
 
@@ -299,12 +378,17 @@ impl PostgresRedoManager {
                 match self.redo_process.get() {
                     None => (),
                     Some(guard) => {
-                        if Arc::ptr_eq(&proc, &*guard) {
-                            // We're the first to observe an error from `proc`, it's our job to take it out of rotation.
-                            guard.take_and_deinit();
-                        } else {
-                            // Another task already spawned another redo process (further up in this method)
-                            // and put it into `redo_process`. Do nothing, our view of the world is behind.
+                        match &*guard {
+                            RedoProcessState::ManagerShutDown => {}
+                            RedoProcessState::Launched(guard_proc) => {
+                                if Arc::ptr_eq(&proc, &*guard_proc) {
+                                    // We're the first to observe an error from `proc`, it's our job to take it out of rotation.
+                                    guard.take_and_deinit();
+                                } else {
+                                    // Another task already spawned another redo process (further up in this method)
+                                    // and put it into `redo_process`. Do nothing, our view of the world is behind.
+                                }
+                            }
                         }
                     }
                 }
@@ -329,7 +413,7 @@ impl PostgresRedoManager {
         lsn: Lsn,
         base_img: Option<Bytes>,
         records: &[(Lsn, NeonWalRecord)],
-    ) -> anyhow::Result<Bytes> {
+    ) -> Result<Bytes, Error> {
         let start_time = Instant::now();
 
         let mut page = BytesMut::new();
@@ -338,7 +422,7 @@ impl PostgresRedoManager {
             page.extend_from_slice(&fpi[..]);
         } else {
             // All the current WAL record types that we can handle require a base image.
-            anyhow::bail!("invalid neon WAL redo request with no base image");
+            bail!("invalid neon WAL redo request with no base image");
         }
 
         // Apply all the WAL records in the batch

pageserver/src/walredo/process.rs

@@ -18,13 +18,21 @@ use std::sync::atomic::AtomicUsize;
 use std::{
     collections::VecDeque,
     process::{Command, Stdio},
+    sync::Arc,
     time::Duration,
 };
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tracing::{debug, error, instrument, Instrument};
-use utils::{lsn::Lsn, poison::Poison};
+use utils::{
+    lsn::Lsn,
+    poison::Poison,
+    sync::gate::{GateError, GateGuard},
+};
+
+use super::GlobalState;
 
 pub struct WalRedoProcess {
+    _spawn_gate_guard: GateGuard,
     #[allow(dead_code)]
     conf: &'static PageServerConf,
     #[cfg(feature = "testing")]
@@ -49,18 +57,33 @@ struct ProcessOutput {
     n_processed_responses: usize,
 }
 
+#[derive(Debug, thiserror::Error)]
+pub(super) enum LaunchError {
+    #[error("cancelled")]
+    Cancelled,
+    #[error(transparent)]
+    Other(#[from] anyhow::Error),
+}
+
 impl WalRedoProcess {
     //
     // Start postgres binary in special WAL redo mode.
     //
     #[instrument(skip_all,fields(pg_version=pg_version))]
     pub(crate) fn launch(
-        conf: &'static PageServerConf,
+        global_state: &Arc<GlobalState>,
         tenant_shard_id: TenantShardId,
         pg_version: u32,
-    ) -> anyhow::Result<Self> {
+    ) -> Result<Self, LaunchError> {
         crate::span::debug_assert_current_span_has_tenant_id();
 
+        let conf = global_state.conf;
+
+        let spawn_gate_guard = match global_state.spawn_gate.enter() {
+            Ok(guard) => guard,
+            Err(GateError::GateClosed) => return Err(LaunchError::Cancelled),
+        };
+
         let pg_bin_dir_path = conf.pg_bin_dir(pg_version).context("pg_bin_dir")?; // TODO these should be infallible.
         let pg_lib_dir_path = conf.pg_lib_dir(pg_version).context("pg_lib_dir")?;
 
@@ -144,6 +167,7 @@ impl WalRedoProcess {
         );
 
         Ok(Self {
+            _spawn_gate_guard: spawn_gate_guard,
             conf,
             #[cfg(feature = "testing")]
             tenant_shard_id,
@@ -189,7 +213,7 @@ impl WalRedoProcess {
         base_img: &Option<Bytes>,
         records: &[(Lsn, NeonWalRecord)],
         wal_redo_timeout: Duration,
-    ) -> anyhow::Result<Bytes> {
+    ) -> Result<Bytes, super::Error> {
         debug_assert_current_span_has_tenant_id();
 
         let tag = protocol::BufferTag { rel, blknum };
@@ -216,17 +240,19 @@ impl WalRedoProcess {
             {
                 protocol::build_apply_record_msg(*lsn, postgres_rec, &mut writebuf);
             } else {
-                anyhow::bail!("tried to pass neon wal record to postgres WAL redo");
+                super::bail!("tried to pass neon wal record to postgres WAL redo");
             }
         }
         protocol::build_get_page_msg(tag, &mut writebuf);
         WAL_REDO_RECORD_COUNTER.inc_by(records.len() as u64);
 
-        let Ok(res) =
-            tokio::time::timeout(wal_redo_timeout, self.apply_wal_records0(&writebuf)).await
-        else {
-            anyhow::bail!("WAL redo timed out");
-        };
+        let res =
+            // TODO: we should tokio::select! on the self.global_state.shutdown here,
+            // but, that requires thinking through the perf implications
+            tokio::time::timeout(wal_redo_timeout, self.apply_wal_records0(&writebuf))
+                .await
+                .map_err(|_elapsed| anyhow::anyhow!("WAL redo timed out"))?
+                .map_err(super::Error::Other);
 
         if res.is_err() {
             // not all of these can be caused by this particular input, however these are so rare
@@ -377,6 +403,11 @@ impl Drop for WalRedoProcess {
             .take()
             .expect("we only do this once")
             .kill_and_wait(WalRedoKillCause::WalRedoProcessDrop);
+
+        // The spawn gate is supposed to track running walredo processes.
+        // => must keep guard alive until the process is dead.
+        let _ = &self._spawn_gate_guard;
+
         // no way to wait for stderr_logger_task from Drop because that is async only
     }
 }

safekeeper/src/wal_storage.rs

@@ -231,7 +231,11 @@ impl PhysicalStorage {
             // half initialized segment, first bake it under tmp filename and
             // then rename.
             let tmp_path = self.timeline_dir.join("waltmp");
-            let mut file = File::create(&tmp_path)
+            #[allow(clippy::suspicious_open_options)]
+            let mut file = OpenOptions::new()
+                .create(true)
+                .write(true)
+                .open(&tmp_path)
                 .await
                 .with_context(|| format!("Failed to open tmp wal file {:?}", &tmp_path))?;
 

test_runner/fixtures/pageserver/http.py

@@ -619,15 +619,13 @@ class PageserverHttpClient(requests.Session, MetricsGetter):
         tenant_id: Union[TenantId, TenantShardId],
         timeline_id: TimelineId,
         timestamp: datetime,
-        with_lease: bool = False,
         **kwargs,
     ):
         log.info(
-            f"Requesting lsn by timestamp {timestamp}, tenant {tenant_id}, timeline {timeline_id}, {with_lease=}"
+            f"Requesting lsn by timestamp {timestamp}, tenant {tenant_id}, timeline {timeline_id}"
         )
-        with_lease_query = f"{with_lease=}".lower()
         res = self.get(
-            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline/{timeline_id}/get_lsn_by_timestamp?timestamp={timestamp.isoformat()}Z&{with_lease_query}",
+            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline/{timeline_id}/get_lsn_by_timestamp?timestamp={timestamp.isoformat()}Z",
             **kwargs,
         )
         self.verbose_error(res)

test_runner/performance/test_storage_controller_scale.py

@@ -48,16 +48,7 @@ def test_storage_controller_many_tenants(
 
     # We will intentionally stress reconciler concurrrency, which triggers a warning when lots
     # of shards are hitting the delayed path.
-    env.storage_controller.allowed_errors.extend(
-        [
-            # We will intentionally stress reconciler concurrrency, which triggers a warning when lots
-            # of shards are hitting the delayed path.
-            ".*Many shards are waiting to reconcile",
-            # We will create many timelines concurrently, so they might get slow enough to trip the warning
-            # that timeline creation is holding a lock too long.
-            ".*Shared lock by TimelineCreate.*was held.*",
-        ]
-    )
+    env.storage_controller.allowed_errors.append(".*Many shards are waiting to reconcile")
 
     for ps in env.pageservers:
         # This can happen because when we do a loop over all pageservers and mark them offline/active,

test_runner/regress/test_lsn_mapping.py

@@ -12,24 +12,10 @@ from fixtures.utils import query_scalar, wait_until
 from requests.exceptions import ReadTimeout
 
 
-def assert_lsn_lease_granted(result, with_lease: bool):
-    """
-    Asserts an LSN lease is granted when `with_lease` flag is turned on.
-    Always asserts no LSN lease is granted when `with_lease` flag is off.
-    """
-    if with_lease:
-        assert result.get("valid_until")
-    else:
-        assert result.get("valid_until") is None
-
-
-@pytest.mark.parametrize("with_lease", [True, False])
-def test_lsn_mapping(neon_env_builder: NeonEnvBuilder, with_lease: bool):
-    """
-    Test pageserver get_lsn_by_timestamp API.
-
-    :param with_lease: Whether to get a lease associated with returned LSN.
-    """
+#
+# Test pageserver get_lsn_by_timestamp API
+#
+def test_lsn_mapping(neon_env_builder: NeonEnvBuilder):
     env = neon_env_builder.init_start()
 
     tenant_id, _ = env.neon_cli.create_tenant(
@@ -81,33 +67,23 @@ def test_lsn_mapping(neon_env_builder: NeonEnvBuilder, with_lease: bool):
         # Check edge cases
         # Timestamp is in the future
         probe_timestamp = tbl[-1][1] + timedelta(hours=1)
-        result = client.timeline_get_lsn_by_timestamp(
-            tenant_id, timeline_id, probe_timestamp, with_lease=with_lease
-        )
+        result = client.timeline_get_lsn_by_timestamp(tenant_id, timeline_id, probe_timestamp)
         assert result["kind"] == "future"
-        assert_lsn_lease_granted(result, with_lease)
         # make sure that we return a well advanced lsn here
         assert Lsn(result["lsn"]) > start_lsn
 
         # Timestamp is in the unreachable past
         probe_timestamp = tbl[0][1] - timedelta(hours=10)
-        result = client.timeline_get_lsn_by_timestamp(
-            tenant_id, timeline_id, probe_timestamp, with_lease=with_lease
-        )
+        result = client.timeline_get_lsn_by_timestamp(tenant_id, timeline_id, probe_timestamp)
         assert result["kind"] == "past"
-        assert_lsn_lease_granted(result, with_lease)
-
         # make sure that we return the minimum lsn here at the start of the range
         assert Lsn(result["lsn"]) < start_lsn
 
         # Probe a bunch of timestamps in the valid range
         for i in range(1, len(tbl), 100):
             probe_timestamp = tbl[i][1]
-            result = client.timeline_get_lsn_by_timestamp(
-                tenant_id, timeline_id, probe_timestamp, with_lease=with_lease
-            )
+            result = client.timeline_get_lsn_by_timestamp(tenant_id, timeline_id, probe_timestamp)
             assert result["kind"] not in ["past", "nodata"]
-            assert_lsn_lease_granted(result, with_lease)
             lsn = result["lsn"]
             # Call get_lsn_by_timestamp to get the LSN
             # Launch a new read-only node at that LSN, and check that only the rows
@@ -129,11 +105,8 @@ def test_lsn_mapping(neon_env_builder: NeonEnvBuilder, with_lease: bool):
 
         # Timestamp is in the unreachable past
         probe_timestamp = tbl[0][1] - timedelta(hours=10)
-        result = client.timeline_get_lsn_by_timestamp(
-            tenant_id, timeline_id_child, probe_timestamp, with_lease=with_lease
-        )
+        result = client.timeline_get_lsn_by_timestamp(tenant_id, timeline_id_child, probe_timestamp)
         assert result["kind"] == "past"
-        assert_lsn_lease_granted(result, with_lease)
         # make sure that we return the minimum lsn here at the start of the range
         assert Lsn(result["lsn"]) >= last_flush_lsn
 

test_runner/regress/test_sharding.py

@@ -190,20 +190,19 @@ def test_sharding_split_compaction(neon_env_builder: NeonEnvBuilder, failpoint:
     """
     Test that after a split, we clean up parent layer data in the child shards via compaction.
     """
-
     TENANT_CONF = {
         # small checkpointing and compaction targets to ensure we generate many upload operations
-        "checkpoint_distance": 128 * 1024,
-        "compaction_threshold": 1,
-        "compaction_target_size": 128 * 1024,
+        "checkpoint_distance": f"{128 * 1024}",
+        "compaction_threshold": "1",
+        "compaction_target_size": f"{128 * 1024}",
         # no PITR horizon, we specify the horizon when we request on-demand GC
         "pitr_interval": "3600s",
         # disable background compaction and GC. We invoke it manually when we want it to happen.
         "gc_period": "0s",
         "compaction_period": "0s",
         # create image layers eagerly, so that GC can remove some layers
-        "image_creation_threshold": 1,
-        "image_layer_creation_check_threshold": 0,
+        "image_creation_threshold": "1",
+        "image_layer_creation_check_threshold": "0",
     }
 
     neon_env_builder.storage_controller_config = {
@@ -262,9 +261,7 @@ def test_sharding_split_compaction(neon_env_builder: NeonEnvBuilder, failpoint:
     env.pageserver.start()
 
     # Cleanup part 2: once layers are outside the PITR window, they will be rewritten if they are partially redundant
-    updated_conf = TENANT_CONF.copy()
-    updated_conf["pitr_interval"] = "0s"
-    env.storage_controller.pageserver_api().set_tenant_config(tenant_id, updated_conf)
+    env.storage_controller.pageserver_api().set_tenant_config(tenant_id, {"pitr_interval": "0s"})
     env.storage_controller.reconcile_until_idle()
 
     for shard in shards:

test_runner/regress/test_tenant_delete.py

@@ -31,12 +31,8 @@ def error_tolerant_delete(ps_http, tenant_id):
             if e.status_code == 500:
                 # This test uses failure injection, which can produce 500s as the pageserver expects
                 # the object store to always be available, and the ListObjects during deletion is generally
-                # an infallible operation.  This can show up as a clear simulated error, or as a general
-                # error during delete_objects()
-                assert (
-                    "simulated failure of remote operation" in e.message
-                    or "failed to delete" in e.message
-                )
+                # an infallible operation
+                assert "simulated failure of remote operation" in e.message
             else:
                 raise
         else: