Add bench feature

This parameter can be use to restrict number of image layers generated
because of GC request (wanted image layers).
Been set to zero it completely eliminates creation of such image layers.
So it allows to avoid extra storage consumption after merging #3673

## Problem
PR #3673 forces generation of missed image layers. So i short term is
cause cause increase (in worst case up to two times) size of storage.
It was intended (by me) that GC period is comparable with PiTR interval.
But looks like it is not the case now - GC is performed much more
frequently. It may cause the problem with space exhaustion: GC forces
new image creation while large PiTR still prevent GC from collecting old
layers.

## Summary of changes

Add new pageserver parameter` forced_image_creation_limit` which
restrict number of created image layers which are requested by GC.

## Checklist before requesting a review

- [ ] I have performed a self-review of my code.
- [ ] If it is a core feature, I have added thorough tests.
- [ ] Do we need to implement analytics? if so did you add the relevant
metrics to the dashboard?
- [ ] If this PR requires public announcement, mark it with
/release-notes label and add several sentences in this section.

## Checklist before merging

- [ ] Do not forget to reformat commit message to not include the above
checklist

control_plane/src/pageserver.rs

@@ -369,6 +369,11 @@ impl PageServerNode {
             evictions_low_residence_duration_metric_threshold: settings
                 .remove("evictions_low_residence_duration_metric_threshold")
                 .map(|x| x.to_string()),
+            gc_feedback: settings
+                .remove("gc_feedback")
+                .map(|x| x.parse::<bool>())
+                .transpose()
+                .context("Failed to parse 'gc_feedback' as bool")?,
         };
 
         // If tenant ID was not specified, generate one
@@ -463,6 +468,11 @@ impl PageServerNode {
                 evictions_low_residence_duration_metric_threshold: settings
                     .remove("evictions_low_residence_duration_metric_threshold")
                     .map(|x| x.to_string()),
+                gc_feedback: settings
+                    .remove("gc_feedback")
+                    .map(|x| x.parse::<bool>())
+                    .transpose()
+                    .context("Failed to parse 'gc_feedback' as bool")?,
             }
         };
 

libs/pageserver_api/src/models.rs

@@ -223,6 +223,7 @@ pub struct TenantConfig {
     pub eviction_policy: Option<serde_json::Value>,
     pub min_resident_size_override: Option<u64>,
     pub evictions_low_residence_duration_metric_threshold: Option<String>,
+    pub gc_feedback: Option<bool>,
 }
 
 #[serde_as]
@@ -281,6 +282,7 @@ impl TenantConfigRequest {
             eviction_policy: None,
             min_resident_size_override: None,
             evictions_low_residence_duration_metric_threshold: None,
+            gc_feedback: None,
         };
         TenantConfigRequest { tenant_id, config }
     }

pageserver/Cargo.toml

@@ -9,6 +9,9 @@ default = []
 # Enables test-only APIs, incuding failpoints. In particular, enables the `fail_point!` macro,
 # which adds some runtime cost to run tests on outage conditions
 testing = ["fail/failpoints"]
+# Just a marker that compiles mock structs that are used in both tests and benchmarks. We
+# hide them behind a feature flag so that we can apply stronger lints to prod-only code.
+bench = []
 
 [dependencies]
 anyhow.workspace = true

pageserver/benches/README.md

@@ -3,10 +3,10 @@
 # How to run
 
 To run all benchmarks:
-`cargo bench`
+`cargo bench --features bench`
 
 To run a specific file:
-`cargo bench --bench bench_layer_map`
+`cargo bench --features bench --bench bench_layer_map`
 
 To run a specific function:
-`cargo bench --bench bench_layer_map -- real_map_uniform_queries`
+`cargo bench --features bench --bench bench_layer_map -- real_map_uniform_queries`

pageserver/benches/bench_layer_map.rs

@@ -1,245 +1,264 @@
-use pageserver::keyspace::{KeyPartitioning, KeySpace};
-use pageserver::repository::Key;
-use pageserver::tenant::layer_map::LayerMap;
-use pageserver::tenant::storage_layer::{Layer, LayerDescriptor, LayerFileName};
-use rand::prelude::{SeedableRng, SliceRandom, StdRng};
-use std::cmp::{max, min};
-use std::fs::File;
-use std::io::{BufRead, BufReader};
-use std::path::PathBuf;
-use std::str::FromStr;
-use std::sync::Arc;
-use std::time::Instant;
+// Hiding this code under a compilation flag allows us to lint it differently than prod code
+#[cfg(feature = "bench")]
+pub mod bench {
+    use pageserver::keyspace::{KeyPartitioning, KeySpace};
+    use pageserver::repository::Key;
+    use pageserver::tenant::layer_map::LayerMap;
+    use pageserver::tenant::storage_layer::mock::LayerDescriptor;
+    use pageserver::tenant::storage_layer::{Layer, LayerFileName};
+    use rand::prelude::{SeedableRng, SliceRandom, StdRng};
+    use std::cmp::{max, min};
+    use std::fs::File;
+    use std::io::{BufRead, BufReader};
+    use std::path::PathBuf;
+    use std::str::FromStr;
+    use std::sync::Arc;
+    use std::time::Instant;
 
-use utils::lsn::Lsn;
+    use utils::lsn::Lsn;
 
-use criterion::{black_box, criterion_group, criterion_main, Criterion};
+    use criterion::{black_box, criterion_group, criterion_main, Criterion};
 
-fn build_layer_map(filename_dump: PathBuf) -> LayerMap<LayerDescriptor> {
-    let mut layer_map = LayerMap::<LayerDescriptor>::default();
+    fn build_layer_map(filename_dump: PathBuf) -> LayerMap<LayerDescriptor> {
+        let mut layer_map = LayerMap::<LayerDescriptor>::default();
 
-    let mut min_lsn = Lsn(u64::MAX);
-    let mut max_lsn = Lsn(0);
+        let mut min_lsn = Lsn(u64::MAX);
+        let mut max_lsn = Lsn(0);
 
-    let filenames = BufReader::new(File::open(filename_dump).unwrap()).lines();
+        let filenames = BufReader::new(File::open(filename_dump).unwrap()).lines();
 
-    let mut updates = layer_map.batch_update();
-    for fname in filenames {
-        let fname = fname.unwrap();
-        let fname = LayerFileName::from_str(&fname).unwrap();
-        let layer = LayerDescriptor::from(fname);
+        let mut updates = layer_map.batch_update();
+        for fname in filenames {
+            let fname = fname.unwrap();
+            let fname = LayerFileName::from_str(&fname).unwrap();
+            let layer = LayerDescriptor::from(fname);
 
-        let lsn_range = layer.get_lsn_range();
-        min_lsn = min(min_lsn, lsn_range.start);
-        max_lsn = max(max_lsn, Lsn(lsn_range.end.0 - 1));
+            let lsn_range = layer.get_lsn_range();
+            min_lsn = min(min_lsn, lsn_range.start);
+            max_lsn = max(max_lsn, Lsn(lsn_range.end.0 - 1));
 
-        updates.insert_historic(Arc::new(layer));
-    }
-
-    println!("min: {min_lsn}, max: {max_lsn}");
-
-    updates.flush();
-    layer_map
-}
-
-/// Construct a layer map query pattern for benchmarks
-fn uniform_query_pattern(layer_map: &LayerMap<LayerDescriptor>) -> Vec<(Key, Lsn)> {
-    // For each image layer we query one of the pages contained, at LSN right
-    // before the image layer was created. This gives us a somewhat uniform
-    // coverage of both the lsn and key space because image layers have
-    // approximately equal sizes and cover approximately equal WAL since
-    // last image.
-    layer_map
-        .iter_historic_layers()
-        .filter_map(|l| {
-            if l.is_incremental() {
-                None
-            } else {
-                let kr = l.get_key_range();
-                let lr = l.get_lsn_range();
-
-                let key_inside = kr.start.next();
-                let lsn_before = Lsn(lr.start.0 - 1);
-
-                Some((key_inside, lsn_before))
-            }
-        })
-        .collect()
-}
-
-// Construct a partitioning for testing get_difficulty map when we
-// don't have an exact result of `collect_keyspace` to work with.
-fn uniform_key_partitioning(layer_map: &LayerMap<LayerDescriptor>, _lsn: Lsn) -> KeyPartitioning {
-    let mut parts = Vec::new();
-
-    // We add a partition boundary at the start of each image layer,
-    // no matter what lsn range it covers. This is just the easiest
-    // thing to do. A better thing to do would be to get a real
-    // partitioning from some database. Even better, remove the need
-    // for key partitions by deciding where to create image layers
-    // directly based on a coverage-based difficulty map.
-    let mut keys: Vec<_> = layer_map
-        .iter_historic_layers()
-        .filter_map(|l| {
-            if l.is_incremental() {
-                None
-            } else {
-                let kr = l.get_key_range();
-                Some(kr.start.next())
-            }
-        })
-        .collect();
-    keys.sort();
-
-    let mut current_key = Key::from_hex("000000000000000000000000000000000000").unwrap();
-    for key in keys {
-        parts.push(KeySpace {
-            ranges: vec![current_key..key],
-        });
-        current_key = key;
-    }
-
-    KeyPartitioning { parts }
-}
-
-// Benchmark using metadata extracted from our performance test environment, from
-// a project where we have run pgbench many timmes. The pgbench database was initialized
-// between each test run.
-fn bench_from_captest_env(c: &mut Criterion) {
-    // TODO consider compressing this file
-    let layer_map = build_layer_map(PathBuf::from("benches/odd-brook-layernames.txt"));
-    let queries: Vec<(Key, Lsn)> = uniform_query_pattern(&layer_map);
-
-    // Test with uniform query pattern
-    c.bench_function("captest_uniform_queries", |b| {
-        b.iter(|| {
-            for q in queries.clone().into_iter() {
-                black_box(layer_map.search(q.0, q.1));
-            }
-        });
-    });
-
-    // test with a key that corresponds to the RelDir entry. See pgdatadir_mapping.rs.
-    c.bench_function("captest_rel_dir_query", |b| {
-        b.iter(|| {
-            let result = black_box(layer_map.search(
-                Key::from_hex("000000067F00008000000000000000000001").unwrap(),
-                // This LSN is higher than any of the LSNs in the tree
-                Lsn::from_str("D0/80208AE1").unwrap(),
-            ));
-            result.unwrap();
-        });
-    });
-}
-
-// Benchmark using metadata extracted from a real project that was taknig
-// too long processing layer map queries.
-fn bench_from_real_project(c: &mut Criterion) {
-    // Init layer map
-    let now = Instant::now();
-    let layer_map = build_layer_map(PathBuf::from("benches/odd-brook-layernames.txt"));
-    println!("Finished layer map init in {:?}", now.elapsed());
-
-    // Choose uniformly distributed queries
-    let queries: Vec<(Key, Lsn)> = uniform_query_pattern(&layer_map);
-
-    // Choose inputs for get_difficulty_map
-    let latest_lsn = layer_map
-        .iter_historic_layers()
-        .map(|l| l.get_lsn_range().end)
-        .max()
-        .unwrap();
-    let partitioning = uniform_key_partitioning(&layer_map, latest_lsn);
-
-    // Check correctness of get_difficulty_map
-    // TODO put this in a dedicated test outside of this mod
-    {
-        println!("running correctness check");
-
-        let now = Instant::now();
-        let result_bruteforce = layer_map.get_difficulty_map_bruteforce(latest_lsn, &partitioning);
-        assert!(result_bruteforce.len() == partitioning.parts.len());
-        println!("Finished bruteforce in {:?}", now.elapsed());
-
-        let now = Instant::now();
-        let result_fast = layer_map.get_difficulty_map(latest_lsn, &partitioning, None);
-        assert!(result_fast.len() == partitioning.parts.len());
-        println!("Finished fast in {:?}", now.elapsed());
-
-        // Assert results are equal. Manually iterate for easier debugging.
-        let zip = std::iter::zip(
-            &partitioning.parts,
-            std::iter::zip(result_bruteforce, result_fast),
-        );
-        for (_part, (bruteforce, fast)) in zip {
-            assert_eq!(bruteforce, fast);
+            updates.insert_historic(Arc::new(layer));
         }
 
-        println!("No issues found");
+        println!("min: {min_lsn}, max: {max_lsn}");
+
+        updates.flush();
+        layer_map
     }
 
-    // Define and name the benchmark function
-    let mut group = c.benchmark_group("real_map");
-    group.bench_function("uniform_queries", |b| {
-        b.iter(|| {
-            for q in queries.clone().into_iter() {
-                black_box(layer_map.search(q.0, q.1));
-            }
-        });
-    });
-    group.bench_function("get_difficulty_map", |b| {
-        b.iter(|| {
-            layer_map.get_difficulty_map(latest_lsn, &partitioning, Some(3));
-        });
-    });
-    group.finish();
-}
+    /// Construct a layer map query pattern for benchmarks
+    fn uniform_query_pattern(layer_map: &LayerMap<LayerDescriptor>) -> Vec<(Key, Lsn)> {
+        // For each image layer we query one of the pages contained, at LSN right
+        // before the image layer was created. This gives us a somewhat uniform
+        // coverage of both the lsn and key space because image layers have
+        // approximately equal sizes and cover approximately equal WAL since
+        // last image.
+        layer_map
+            .iter_historic_layers()
+            .filter_map(|l| {
+                if l.is_incremental() {
+                    None
+                } else {
+                    let kr = l.get_key_range();
+                    let lr = l.get_lsn_range();
 
-// Benchmark using synthetic data. Arrange image layers on stacked diagonal lines.
-fn bench_sequential(c: &mut Criterion) {
-    // Init layer map. Create 100_000 layers arranged in 1000 diagonal lines.
-    //
-    // TODO This code is pretty slow and runs even if we're only running other
-    //      benchmarks. It needs to be somewhere else, but it's not clear where.
-    //      Putting it inside the `bench_function` closure is not a solution
-    //      because then it runs multiple times during warmup.
-    let now = Instant::now();
-    let mut layer_map = LayerMap::default();
-    let mut updates = layer_map.batch_update();
-    for i in 0..100_000 {
-        let i32 = (i as u32) % 100;
-        let zero = Key::from_hex("000000000000000000000000000000000000").unwrap();
-        let layer = LayerDescriptor {
-            key: zero.add(10 * i32)..zero.add(10 * i32 + 1),
-            lsn: Lsn(i)..Lsn(i + 1),
-            is_incremental: false,
-            short_id: format!("Layer {}", i),
-        };
-        updates.insert_historic(Arc::new(layer));
+                    let key_inside = kr.start.next();
+                    let lsn_before = Lsn(lr.start.0 - 1);
+
+                    Some((key_inside, lsn_before))
+                }
+            })
+            .collect()
     }
-    updates.flush();
-    println!("Finished layer map init in {:?}", now.elapsed());
 
-    // Choose 100 uniformly random queries
-    let rng = &mut StdRng::seed_from_u64(1);
-    let queries: Vec<(Key, Lsn)> = uniform_query_pattern(&layer_map)
-        .choose_multiple(rng, 100)
-        .copied()
-        .collect();
+    // Construct a partitioning for testing get_difficulty map when we
+    // don't have an exact result of `collect_keyspace` to work with.
+    fn uniform_key_partitioning(
+        layer_map: &LayerMap<LayerDescriptor>,
+        _lsn: Lsn,
+    ) -> KeyPartitioning {
+        let mut parts = Vec::new();
 
-    // Define and name the benchmark function
-    let mut group = c.benchmark_group("sequential");
-    group.bench_function("uniform_queries", |b| {
-        b.iter(|| {
-            for q in queries.clone().into_iter() {
-                black_box(layer_map.search(q.0, q.1));
-            }
+        // We add a partition boundary at the start of each image layer,
+        // no matter what lsn range it covers. This is just the easiest
+        // thing to do. A better thing to do would be to get a real
+        // partitioning from some database. Even better, remove the need
+        // for key partitions by deciding where to create image layers
+        // directly based on a coverage-based difficulty map.
+        let mut keys: Vec<_> = layer_map
+            .iter_historic_layers()
+            .filter_map(|l| {
+                if l.is_incremental() {
+                    None
+                } else {
+                    let kr = l.get_key_range();
+                    Some(kr.start.next())
+                }
+            })
+            .collect();
+        keys.sort();
+
+        let mut current_key = Key::from_hex("000000000000000000000000000000000000").unwrap();
+        for key in keys {
+            parts.push(KeySpace {
+                ranges: vec![current_key..key],
+            });
+            current_key = key;
+        }
+
+        KeyPartitioning { parts }
+    }
+
+    // Benchmark using metadata extracted from our performance test environment, from
+    // a project where we have run pgbench many timmes. The pgbench database was initialized
+    // between each test run.
+    fn bench_from_captest_env(c: &mut Criterion) {
+        // TODO consider compressing this file
+        let layer_map = build_layer_map(PathBuf::from("benches/odd-brook-layernames.txt"));
+        let queries: Vec<(Key, Lsn)> = uniform_query_pattern(&layer_map);
+
+        // Test with uniform query pattern
+        c.bench_function("captest_uniform_queries", |b| {
+            b.iter(|| {
+                for q in queries.clone().into_iter() {
+                    black_box(layer_map.search(q.0, q.1));
+                }
+            });
         });
-    });
-    group.finish();
+
+        // test with a key that corresponds to the RelDir entry. See pgdatadir_mapping.rs.
+        c.bench_function("captest_rel_dir_query", |b| {
+            b.iter(|| {
+                let result = black_box(layer_map.search(
+                    Key::from_hex("000000067F00008000000000000000000001").unwrap(),
+                    // This LSN is higher than any of the LSNs in the tree
+                    Lsn::from_str("D0/80208AE1").unwrap(),
+                ));
+                result.unwrap();
+            });
+        });
+    }
+
+    // Benchmark using metadata extracted from a real project that was taknig
+    // too long processing layer map queries.
+    fn bench_from_real_project(c: &mut Criterion) {
+        // Init layer map
+        let now = Instant::now();
+        let layer_map = build_layer_map(PathBuf::from("benches/odd-brook-layernames.txt"));
+        println!("Finished layer map init in {:?}", now.elapsed());
+
+        // Choose uniformly distributed queries
+        let queries: Vec<(Key, Lsn)> = uniform_query_pattern(&layer_map);
+
+        // Choose inputs for get_difficulty_map
+        let latest_lsn = layer_map
+            .iter_historic_layers()
+            .map(|l| l.get_lsn_range().end)
+            .max()
+            .unwrap();
+        let partitioning = uniform_key_partitioning(&layer_map, latest_lsn);
+
+        // Check correctness of get_difficulty_map
+        // TODO put this in a dedicated test outside of this mod
+        {
+            println!("running correctness check");
+
+            let now = Instant::now();
+            let result_bruteforce =
+                layer_map.get_difficulty_map_bruteforce(latest_lsn, &partitioning);
+            assert!(result_bruteforce.len() == partitioning.parts.len());
+            println!("Finished bruteforce in {:?}", now.elapsed());
+
+            let now = Instant::now();
+            let result_fast = layer_map.get_difficulty_map(latest_lsn, &partitioning, None);
+            assert!(result_fast.len() == partitioning.parts.len());
+            println!("Finished fast in {:?}", now.elapsed());
+
+            // Assert results are equal. Manually iterate for easier debugging.
+            let zip = std::iter::zip(
+                &partitioning.parts,
+                std::iter::zip(result_bruteforce, result_fast),
+            );
+            for (_part, (bruteforce, fast)) in zip {
+                assert_eq!(bruteforce, fast);
+            }
+
+            println!("No issues found");
+        }
+
+        // Define and name the benchmark function
+        let mut group = c.benchmark_group("real_map");
+        group.bench_function("uniform_queries", |b| {
+            b.iter(|| {
+                for q in queries.clone().into_iter() {
+                    black_box(layer_map.search(q.0, q.1));
+                }
+            });
+        });
+        group.bench_function("get_difficulty_map", |b| {
+            b.iter(|| {
+                layer_map.get_difficulty_map(latest_lsn, &partitioning, Some(3));
+            });
+        });
+        group.finish();
+    }
+
+    // Benchmark using synthetic data. Arrange image layers on stacked diagonal lines.
+    fn bench_sequential(c: &mut Criterion) {
+        // Init layer map. Create 100_000 layers arranged in 1000 diagonal lines.
+        //
+        // TODO This code is pretty slow and runs even if we're only running other
+        //      benchmarks. It needs to be somewhere else, but it's not clear where.
+        //      Putting it inside the `bench_function` closure is not a solution
+        //      because then it runs multiple times during warmup.
+        let now = Instant::now();
+        let mut layer_map = LayerMap::default();
+        let mut updates = layer_map.batch_update();
+        for i in 0..100_000 {
+            let i32 = (i as u32) % 100;
+            let zero = Key::from_hex("000000000000000000000000000000000000").unwrap();
+            let layer = LayerDescriptor {
+                key: zero.add(10 * i32)..zero.add(10 * i32 + 1),
+                lsn: Lsn(i)..Lsn(i + 1),
+                is_incremental: false,
+                short_id: format!("Layer {}", i),
+            };
+            updates.insert_historic(Arc::new(layer));
+        }
+        updates.flush();
+        println!("Finished layer map init in {:?}", now.elapsed());
+
+        // Choose 100 uniformly random queries
+        let rng = &mut StdRng::seed_from_u64(1);
+        let queries: Vec<(Key, Lsn)> = uniform_query_pattern(&layer_map)
+            .choose_multiple(rng, 100)
+            .copied()
+            .collect();
+
+        // Define and name the benchmark function
+        let mut group = c.benchmark_group("sequential");
+        group.bench_function("uniform_queries", |b| {
+            b.iter(|| {
+                for q in queries.clone().into_iter() {
+                    black_box(layer_map.search(q.0, q.1));
+                }
+            });
+        });
+        group.finish();
+    }
+
+    criterion_group!(group_1, bench_from_captest_env);
+    criterion_group!(group_2, bench_from_real_project);
+    criterion_group!(group_3, bench_sequential);
 }
 
-criterion_group!(group_1, bench_from_captest_env);
-criterion_group!(group_2, bench_from_real_project);
-criterion_group!(group_3, bench_sequential);
-criterion_main!(group_1, group_2, group_3);
+#[cfg(feature = "bench")]
+use criterion::criterion_main;
+
+#[cfg(feature = "bench")]
+criterion_main!(bench::group_1, bench::group_2, bench::group_3);
+
+#[cfg(not(feature = "bench"))]
+fn main() {
+    panic!("Use `--features bench` to run benchmarks")
+}

pageserver/src/config.rs

@@ -108,7 +108,7 @@ pub mod defaults {
 
 #min_resident_size_override = .. # in bytes
 #evictions_low_residence_duration_metric_threshold = '{DEFAULT_EVICTIONS_LOW_RESIDENCE_DURATION_METRIC_THRESHOLD}'
-
+#gc_feedback = false
 # [remote_storage]
 
 "###
@@ -828,6 +828,14 @@ impl PageServerConf {
             )?);
         }
 
+        if let Some(gc_feedback) = item.get("gc_feedback") {
+            t_conf.gc_feedback = Some(
+                gc_feedback
+                    .as_bool()
+                    .with_context(|| "configure option gc_feedback is not a bool".to_string())?,
+            );
+        }
+
         Ok(t_conf)
     }
 

pageserver/src/tenant.rs

@@ -3209,6 +3209,7 @@ pub mod harness {
                 evictions_low_residence_duration_metric_threshold: Some(
                     tenant_conf.evictions_low_residence_duration_metric_threshold,
                 ),
+                gc_feedback: Some(tenant_conf.gc_feedback),
             }
         }
     }

pageserver/src/tenant/config.rs

@@ -99,6 +99,7 @@ pub struct TenantConf {
     // See the corresponding metric's help string.
     #[serde(with = "humantime_serde")]
     pub evictions_low_residence_duration_metric_threshold: Duration,
+    pub gc_feedback: bool,
 }
 
 /// Same as TenantConf, but this struct preserves the information about
@@ -175,6 +176,10 @@ pub struct TenantConfOpt {
     #[serde(with = "humantime_serde")]
     #[serde(default)]
     pub evictions_low_residence_duration_metric_threshold: Option<Duration>,
+
+    #[serde(skip_serializing_if = "Option::is_none")]
+    #[serde(default)]
+    pub gc_feedback: Option<bool>,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
@@ -242,6 +247,7 @@ impl TenantConfOpt {
             evictions_low_residence_duration_metric_threshold: self
                 .evictions_low_residence_duration_metric_threshold
                 .unwrap_or(global_conf.evictions_low_residence_duration_metric_threshold),
+            gc_feedback: self.gc_feedback.unwrap_or(global_conf.gc_feedback),
         }
     }
 }
@@ -278,6 +284,7 @@ impl Default for TenantConf {
                 DEFAULT_EVICTIONS_LOW_RESIDENCE_DURATION_METRIC_THRESHOLD,
             )
             .expect("cannot parse default evictions_low_residence_duration_metric_threshold"),
+            gc_feedback: false,
         }
     }
 }
@@ -372,6 +379,7 @@ impl TryFrom<&'_ models::TenantConfig> for TenantConfOpt {
                     ))?,
             );
         }
+        tenant_conf.gc_feedback = request_data.gc_feedback;
 
         Ok(tenant_conf)
     }

pageserver/src/tenant/layer_map.rs

@@ -762,7 +762,8 @@ where
 #[cfg(test)]
 mod tests {
     use super::{LayerMap, Replacement};
-    use crate::tenant::storage_layer::{Layer, LayerDescriptor, LayerFileName};
+    use crate::tenant::storage_layer::mock::LayerDescriptor;
+    use crate::tenant::storage_layer::{Layer, LayerFileName};
     use std::str::FromStr;
     use std::sync::Arc;
 

pageserver/src/tenant/layer_map/historic_layer_coverage.rs

@@ -204,6 +204,35 @@ fn test_off_by_one() {
     assert_eq!(version.image_coverage.query(5), None);
 }
 
+/// White-box regression test, checking for incorrect removal of node at key.end
+#[test]
+fn test_regression() {
+    let mut map = HistoricLayerCoverage::<String>::new();
+    map.insert(
+        LayerKey {
+            key: 0..5,
+            lsn: 0..5,
+            is_image: false,
+        },
+        "Layer 1".to_string(),
+    );
+    map.insert(
+        LayerKey {
+            key: 0..5,
+            lsn: 1..2,
+            is_image: false,
+        },
+        "Layer 2".to_string(),
+    );
+
+    // If an insertion operation improperly deletes the endpoint of a previous layer
+    // (which is more likely to happen with layers that collide on key.end), we will
+    // end up with an infinite layer, covering the entire keyspace. Here we assert
+    // that there's no layer at key 100 because we didn't insert any layer there.
+    let version = map.get_version(100).unwrap();
+    assert_eq!(version.delta_coverage.query(100), None);
+}
+
 /// Cover edge cases where layers begin or end on the same key
 #[test]
 fn test_key_collision() {

pageserver/src/tenant/layer_map/layer_coverage.rs

@@ -10,19 +10,22 @@ use rpds::RedBlackTreeMapSync;
 /// - iterate the latest layers in a key range
 /// - insert layers in non-decreasing lsn.start order
 ///
-/// The struct is parameterized over Value for easier
-/// testing, but in practice it's some sort of layer.
+/// For a detailed explanation and justification of this approach, see:
+/// https://neon.tech/blog/persistent-structures-in-neons-wal-indexing
+///
+/// NOTE The struct is parameterized over Value for easier
+///      testing, but in practice it's some sort of layer.
 pub struct LayerCoverage<Value> {
     /// For every change in coverage (as we sweep the key space)
     /// we store (lsn.end, value).
     ///
-    /// We use an immutable/persistent tree so that we can keep historic
-    /// versions of this coverage without cloning the whole thing and
-    /// incurring quadratic memory cost. See HistoricLayerCoverage.
+    /// NOTE We use an immutable/persistent tree so that we can keep historic
+    ///      versions of this coverage without cloning the whole thing and
+    ///      incurring quadratic memory cost. See HistoricLayerCoverage.
     ///
-    /// We use the Sync version of the map because we want Self to
-    /// be Sync. Using nonsync might be faster, if we can work with
-    /// that.
+    /// NOTE We use the Sync version of the map because we want Self to
+    ///      be Sync. Using nonsync might be faster, if we can work with
+    ///      that.
     nodes: RedBlackTreeMapSync<i128, Option<(u64, Value)>>,
 }
 
@@ -41,6 +44,13 @@ impl<Value: Clone> LayerCoverage<Value> {
 
     /// Helper function to subdivide the key range without changing any values
     ///
+    /// This operation has no semantic effect by itself. It only helps us pin in
+    /// place the part of the coverage we don't want to change when inserting.
+    ///
+    /// As an analogy, think of a polygon. If you add a vertex along one of the
+    /// segments, the polygon is still the same, but it behaves differently when
+    /// we move or delete one of the other points.
+    ///
     /// Complexity: O(log N)
     fn add_node(&mut self, key: i128) {
         let value = match self.nodes.range(..=key).last() {
@@ -74,7 +84,7 @@ impl<Value: Clone> LayerCoverage<Value> {
         let mut to_update = Vec::new();
         let mut to_remove = Vec::new();
         let mut prev_covered = false;
-        for (k, node) in self.nodes.range(key.clone()) {
+        for (k, node) in self.nodes.range(key) {
             let needs_cover = match node {
                 None => true,
                 Some((h, _)) => h < &lsn.end,
@@ -87,9 +97,8 @@ impl<Value: Clone> LayerCoverage<Value> {
             }
             prev_covered = needs_cover;
         }
-        if !prev_covered {
-            to_remove.push(key.end);
-        }
+        // TODO check if the nodes inserted at key.start and key.end are safe
+        //      to remove. It's fine to keep them but they could be redundant.
         for k in to_update {
             self.nodes.insert_mut(k, Some((lsn.end, value.clone())));
         }

pageserver/src/tenant/storage_layer.rs

@@ -460,80 +460,86 @@ pub fn downcast_remote_layer(
     }
 }
 
-/// Holds metadata about a layer without any content. Used mostly for testing.
-///
-/// To use filenames as fixtures, parse them as [`LayerFileName`] then convert from that to a
-/// LayerDescriptor.
-#[derive(Clone, Debug)]
-pub struct LayerDescriptor {
-    pub key: Range<Key>,
-    pub lsn: Range<Lsn>,
-    pub is_incremental: bool,
-    pub short_id: String,
-}
+// Hiding this code under a compilation flag allows us to lint it differently than prod code
+#[cfg(any(test, feature = "bench"))]
+pub mod mock {
+    use super::*;
 
-impl Layer for LayerDescriptor {
-    fn get_key_range(&self) -> Range<Key> {
-        self.key.clone()
+    /// Holds metadata about a layer without any content. Used mostly for testing.
+    ///
+    /// To use filenames as fixtures, parse them as [`LayerFileName`] then convert from that to a
+    /// LayerDescriptor.
+    #[derive(Clone, Debug)]
+    pub struct LayerDescriptor {
+        pub key: Range<Key>,
+        pub lsn: Range<Lsn>,
+        pub is_incremental: bool,
+        pub short_id: String,
     }
 
-    fn get_lsn_range(&self) -> Range<Lsn> {
-        self.lsn.clone()
-    }
+    impl Layer for LayerDescriptor {
+        fn get_key_range(&self) -> Range<Key> {
+            self.key.clone()
+        }
 
-    fn is_incremental(&self) -> bool {
-        self.is_incremental
-    }
+        fn get_lsn_range(&self) -> Range<Lsn> {
+            self.lsn.clone()
+        }
 
-    fn get_value_reconstruct_data(
-        &self,
-        _key: Key,
-        _lsn_range: Range<Lsn>,
-        _reconstruct_data: &mut ValueReconstructState,
-        _ctx: &RequestContext,
-    ) -> Result<ValueReconstructResult> {
-        todo!("This method shouldn't be part of the Layer trait")
-    }
+        fn is_incremental(&self) -> bool {
+            self.is_incremental
+        }
 
-    fn short_id(&self) -> String {
-        self.short_id.clone()
-    }
+        fn get_value_reconstruct_data(
+            &self,
+            _key: Key,
+            _lsn_range: Range<Lsn>,
+            _reconstruct_data: &mut ValueReconstructState,
+            _ctx: &RequestContext,
+        ) -> Result<ValueReconstructResult> {
+            todo!("This method shouldn't be part of the Layer trait")
+        }
 
-    fn dump(&self, _verbose: bool, _ctx: &RequestContext) -> Result<()> {
-        todo!()
-    }
-}
+        fn short_id(&self) -> String {
+            self.short_id.clone()
+        }
 
-impl From<DeltaFileName> for LayerDescriptor {
-    fn from(value: DeltaFileName) -> Self {
-        let short_id = value.to_string();
-        LayerDescriptor {
-            key: value.key_range,
-            lsn: value.lsn_range,
-            is_incremental: true,
-            short_id,
+        fn dump(&self, _verbose: bool, _ctx: &RequestContext) -> Result<()> {
+            todo!()
         }
     }
-}
 
-impl From<ImageFileName> for LayerDescriptor {
-    fn from(value: ImageFileName) -> Self {
-        let short_id = value.to_string();
-        let lsn = value.lsn_as_range();
-        LayerDescriptor {
-            key: value.key_range,
-            lsn,
-            is_incremental: false,
-            short_id,
+    impl From<DeltaFileName> for LayerDescriptor {
+        fn from(value: DeltaFileName) -> Self {
+            let short_id = value.to_string();
+            LayerDescriptor {
+                key: value.key_range,
+                lsn: value.lsn_range,
+                is_incremental: true,
+                short_id,
+            }
         }
     }
-}
 
-impl From<LayerFileName> for LayerDescriptor {
-    fn from(value: LayerFileName) -> Self {
-        match value {
-            LayerFileName::Delta(d) => Self::from(d),
-            LayerFileName::Image(i) => Self::from(i),
+    impl From<ImageFileName> for LayerDescriptor {
+        fn from(value: ImageFileName) -> Self {
+            let short_id = value.to_string();
+            let lsn = value.lsn_as_range();
+            LayerDescriptor {
+                key: value.key_range,
+                lsn,
+                is_incremental: false,
+                short_id,
+            }
+        }
+    }
+
+    impl From<LayerFileName> for LayerDescriptor {
+        fn from(value: LayerFileName) -> Self {
+            match value {
+                LayerFileName::Delta(d) => Self::from(d),
+                LayerFileName::Image(i) => Self::from(i),
+            }
         }
     }
 }

pageserver/src/tenant/timeline.rs

@@ -1291,6 +1291,13 @@ impl Timeline {
             .unwrap_or(default_tenant_conf.evictions_low_residence_duration_metric_threshold)
     }
 
+    fn get_gc_feedback(&self) -> bool {
+        let tenant_conf = self.tenant_conf.read().unwrap();
+        tenant_conf
+            .gc_feedback
+            .unwrap_or(self.conf.default_tenant_conf.gc_feedback)
+    }
+
     pub(super) fn tenant_conf_updated(&self) {
         // NB: Most tenant conf options are read by background loops, so,
         // changes will automatically be picked up.
@@ -3124,6 +3131,7 @@ impl Timeline {
         let mut layers = self.layers.write().unwrap();
         let mut updates = layers.batch_update();
         let timeline_path = self.conf.timeline_path(&self.timeline_id, &self.tenant_id);
+
         for l in image_layers {
             let path = l.filename();
             let metadata = timeline_path
@@ -3896,7 +3904,7 @@ impl Timeline {
                 // delta layers. Image layers can form "stairs" preventing old image from been deleted.
                 // But image layers are in any case less sparse than delta layers. Also we need some
                 // protection from replacing recent image layers with new one after each GC iteration.
-                if l.is_incremental() && !LayerMap::is_l0(&*l) {
+                if self.get_gc_feedback() && l.is_incremental() && !LayerMap::is_l0(&*l) {
                     wanted_image_layers.add_range(l.get_key_range());
                 }
                 result.layers_not_updated += 1;

test_runner/fixtures/neon_fixtures.py

@@ -1621,6 +1621,8 @@ class NeonPageserver(PgProtocol):
             ".*Compaction failed, retrying in [^:]+: Cannot run compaction iteration on inactive tenant",
             # these can happen anytime we do compactions from background task and shutdown pageserver
             r".*ERROR.*ancestor timeline \S+ is being stopped",
+            # this is expected given our collaborative shutdown approach for the UploadQueue
+            ".*Compaction failed, retrying in .*: queue is in state Stopped.*",
         ]
 
     def start(

test_runner/regress/test_attach_tenant_config.py

@@ -158,6 +158,7 @@ def test_fully_custom_config(positive_env: NeonEnv):
             "threshold": "23h",
         },
         "evictions_low_residence_duration_metric_threshold": "2days",
+        "gc_feedback": True,
         "gc_horizon": 23 * (1024 * 1024),
         "gc_period": "2h 13m",
         "image_creation_threshold": 7,