Log detection of duplicate layer

Before finishing the delta file, and possibly overwriting an old
perfectly valid file, check if an identical file already exists in the
layer map.

This is an alternative for https://github.com/neondatabase/neon/pull/4094.
Test case is copied from that PR.

pageserver/src/keyspace.rs

@@ -5,7 +5,7 @@ use std::ops::Range;
 ///
 /// Represents a set of Keys, in a compact form.
 ///
-#[derive(Clone, Debug, Default)]
+#[derive(Clone, Debug)]
 pub struct KeySpace {
     /// Contiguous ranges of keys that belong to the key space. In key order,
     /// and with no overlap.
@@ -61,58 +61,6 @@ impl KeySpace {
 
         KeyPartitioning { parts }
     }
-
-    ///
-    /// Calculate logical size of delta layers: total size of all blocks covered by it's key range
-    ///
-    pub fn get_logical_size(&self, range: &Range<Key>) -> u64 {
-        let mut start_key = range.start;
-        let n_ranges = self.ranges.len();
-        let start_index = match self.ranges.binary_search_by_key(&start_key, |r| r.start) {
-            Ok(index) => index, // keyspace range starts with start_key
-            Err(index) => {
-                if index != 0 && self.ranges[index - 1].end > start_key {
-                    index - 1 // previous keyspace range overlaps with specified
-                } else if index == n_ranges {
-                    return 0; // no intersection with specified range
-                } else {
-                    start_key = self.ranges[index].start;
-                    index
-                }
-            }
-        };
-        let mut size = 0u64;
-        for i in start_index..n_ranges {
-            if self.ranges[i].start >= range.end {
-                break;
-            }
-            let end_key = if self.ranges[i].end < range.end {
-                self.ranges[i].end
-            } else {
-                range.end
-            };
-            let n_blocks = key_range_size(&(start_key..end_key));
-            if n_blocks != u32::MAX {
-                size += n_blocks as u64 * BLCKSZ as u64;
-            }
-            if i + 1 < n_ranges {
-                start_key = self.ranges[i + 1].start;
-            }
-        }
-        size
-    }
-
-    ///
-    /// Check if key space contains overlapping range
-    ///
-    pub fn overlaps(&self, range: &Range<Key>) -> bool {
-        match self.ranges.binary_search_by_key(&range.end, |r| r.start) {
-            Ok(0) => false,
-            Err(0) => false,
-            Ok(index) => self.ranges[index - 1].end > range.start,
-            Err(index) => self.ranges[index - 1].end > range.start,
-        }
-    }
 }
 
 ///
@@ -181,226 +129,3 @@ impl KeySpaceAccum {
         }
     }
 }
-
-///
-/// A helper object, to collect a set of keys and key ranges into a KeySpace
-/// object. Key ranges may be inserted in any order and can overlap.
-///
-#[derive(Clone, Debug, Default)]
-pub struct KeySpaceRandomAccum {
-    ranges: Vec<Range<Key>>,
-}
-
-impl KeySpaceRandomAccum {
-    pub fn new() -> Self {
-        Self { ranges: Vec::new() }
-    }
-
-    pub fn add_key(&mut self, key: Key) {
-        self.add_range(singleton_range(key))
-    }
-
-    pub fn add_range(&mut self, range: Range<Key>) {
-        self.ranges.push(range);
-    }
-
-    pub fn to_keyspace(mut self) -> KeySpace {
-        let mut ranges = Vec::new();
-        if !self.ranges.is_empty() {
-            self.ranges.sort_by_key(|r| r.start);
-            let mut start = self.ranges.first().unwrap().start;
-            let mut end = self.ranges.first().unwrap().end;
-            for r in self.ranges {
-                assert!(r.start >= start);
-                if r.start > end {
-                    ranges.push(start..end);
-                    start = r.start;
-                    end = r.end;
-                } else if r.end > end {
-                    end = r.end;
-                }
-            }
-            ranges.push(start..end);
-        }
-        KeySpace { ranges }
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use std::fmt::Write;
-
-    // Helper function to create a key range.
-    //
-    // Make the tests below less verbose.
-    fn kr(irange: Range<i128>) -> Range<Key> {
-        Key::from_i128(irange.start)..Key::from_i128(irange.end)
-    }
-
-    #[allow(dead_code)]
-    fn dump_keyspace(ks: &KeySpace) {
-        for r in ks.ranges.iter() {
-            println!("  {}..{}", r.start.to_i128(), r.end.to_i128());
-        }
-    }
-
-    fn assert_ks_eq(actual: &KeySpace, expected: Vec<Range<Key>>) {
-        if actual.ranges != expected {
-            let mut msg = String::new();
-
-            writeln!(msg, "expected:").unwrap();
-            for r in &expected {
-                writeln!(msg, "  {}..{}", r.start.to_i128(), r.end.to_i128()).unwrap();
-            }
-            writeln!(msg, "got:").unwrap();
-            for r in &actual.ranges {
-                writeln!(msg, "  {}..{}", r.start.to_i128(), r.end.to_i128()).unwrap();
-            }
-            panic!("{}", msg);
-        }
-    }
-
-    #[test]
-    fn keyspace_add_range() {
-        // two separate ranges
-        //
-        // #####
-        //         #####
-        let mut ks = KeySpaceRandomAccum::default();
-        ks.add_range(kr(0..10));
-        ks.add_range(kr(20..30));
-        assert_ks_eq(&ks.to_keyspace(), vec![kr(0..10), kr(20..30)]);
-
-        // two separate ranges, added in reverse order
-        //
-        //         #####
-        // #####
-        let mut ks = KeySpaceRandomAccum::default();
-        ks.add_range(kr(20..30));
-        ks.add_range(kr(0..10));
-
-        // add range that is adjacent to the end of an existing range
-        //
-        // #####
-        //      #####
-        ks.add_range(kr(0..10));
-        ks.add_range(kr(10..30));
-        assert_ks_eq(&ks.to_keyspace(), vec![kr(0..30)]);
-
-        // add range that is adjacent to the start of an existing range
-        //
-        //      #####
-        // #####
-        let mut ks = KeySpaceRandomAccum::default();
-        ks.add_range(kr(10..30));
-        ks.add_range(kr(0..10));
-        assert_ks_eq(&ks.to_keyspace(), vec![kr(0..30)]);
-
-        // add range that overlaps with the end of an existing range
-        //
-        // #####
-        //    #####
-        let mut ks = KeySpaceRandomAccum::default();
-        ks.add_range(kr(0..10));
-        ks.add_range(kr(5..30));
-        assert_ks_eq(&ks.to_keyspace(), vec![kr(0..30)]);
-
-        // add range that overlaps with the start of an existing range
-        //
-        //    #####
-        // #####
-        let mut ks = KeySpaceRandomAccum::default();
-        ks.add_range(kr(5..30));
-        ks.add_range(kr(0..10));
-        assert_ks_eq(&ks.to_keyspace(), vec![kr(0..30)]);
-
-        // add range that is fully covered by an existing range
-        //
-        // #########
-        //   #####
-        let mut ks = KeySpaceRandomAccum::default();
-        ks.add_range(kr(0..30));
-        ks.add_range(kr(10..20));
-        assert_ks_eq(&ks.to_keyspace(), vec![kr(0..30)]);
-
-        // add range that extends an existing range from both ends
-        //
-        //   #####
-        // #########
-        let mut ks = KeySpaceRandomAccum::default();
-        ks.add_range(kr(10..20));
-        ks.add_range(kr(0..30));
-        assert_ks_eq(&ks.to_keyspace(), vec![kr(0..30)]);
-
-        // add a range that overlaps with two existing ranges, joining them
-        //
-        // #####   #####
-        //    #######
-        let mut ks = KeySpaceRandomAccum::default();
-        ks.add_range(kr(0..10));
-        ks.add_range(kr(20..30));
-        ks.add_range(kr(5..25));
-        assert_ks_eq(&ks.to_keyspace(), vec![kr(0..30)]);
-    }
-
-    #[test]
-    fn keyspace_overlaps() {
-        let mut ks = KeySpaceRandomAccum::default();
-        ks.add_range(kr(10..20));
-        ks.add_range(kr(30..40));
-        let ks = ks.to_keyspace();
-
-        //        #####      #####
-        // xxxx
-        assert!(!ks.overlaps(&kr(0..5)));
-
-        //        #####      #####
-        //   xxxx
-        assert!(!ks.overlaps(&kr(5..9)));
-
-        //        #####      #####
-        //    xxxx
-        assert!(!ks.overlaps(&kr(5..10)));
-
-        //        #####      #####
-        //     xxxx
-        assert!(ks.overlaps(&kr(5..11)));
-
-        //        #####      #####
-        //        xxxx
-        assert!(ks.overlaps(&kr(10..15)));
-
-        //        #####      #####
-        //         xxxx
-        assert!(ks.overlaps(&kr(15..20)));
-
-        //        #####      #####
-        //           xxxx
-        assert!(ks.overlaps(&kr(15..25)));
-
-        //        #####      #####
-        //              xxxx
-        assert!(!ks.overlaps(&kr(22..28)));
-
-        //        #####      #####
-        //               xxxx
-        assert!(!ks.overlaps(&kr(25..30)));
-
-        //        #####      #####
-        //                      xxxx
-        assert!(ks.overlaps(&kr(35..35)));
-
-        //        #####      #####
-        //                        xxxx
-        assert!(!ks.overlaps(&kr(40..45)));
-
-        //        #####      #####
-        //                        xxxx
-        assert!(!ks.overlaps(&kr(45..50)));
-
-        //        #####      #####
-        //        xxxxxxxxxxx
-        assert!(ks.overlaps(&kr(0..30))); // XXXXX This fails currently!
-    }
-}

pageserver/src/tenant.rs

@@ -2195,7 +2195,7 @@ impl Tenant {
                 // made.
                 break;
             }
-            let result = timeline.gc(ctx).await?;
+            let result = timeline.gc().await?;
             totals += result;
         }
 
@@ -3754,7 +3754,7 @@ mod tests {
                 .await?;
             tline.freeze_and_flush().await?;
             tline.compact(&ctx).await?;
-            tline.gc(&ctx).await?;
+            tline.gc().await?;
         }
 
         Ok(())
@@ -3826,7 +3826,7 @@ mod tests {
                 .await?;
             tline.freeze_and_flush().await?;
             tline.compact(&ctx).await?;
-            tline.gc(&ctx).await?;
+            tline.gc().await?;
         }
 
         Ok(())
@@ -3910,7 +3910,7 @@ mod tests {
                 .await?;
             tline.freeze_and_flush().await?;
             tline.compact(&ctx).await?;
-            tline.gc(&ctx).await?;
+            tline.gc().await?;
         }
 
         Ok(())

pageserver/src/tenant/layer_map.rs

@@ -285,6 +285,11 @@ where
         }
     }
 
+    pub fn contains(&self, key_range: &Range<Key>, lsn_range: &Range<Lsn>, is_image: bool) -> bool {
+        let key = historic_layer_coverage::LayerKey::from_ranges(key_range, lsn_range, is_image);
+        self.historic.contains(&key)
+    }
+
     ///
     /// Remove an on-disk layer from the map.
     ///

pageserver/src/tenant/layer_map/historic_layer_coverage.rs

@@ -53,6 +53,20 @@ impl<'a, L: crate::tenant::storage_layer::Layer + ?Sized> From<&'a L> for LayerK
     }
 }
 
+impl LayerKey {
+    pub fn from_ranges(
+        kr: &Range<crate::tenant::layer_map::Key>,
+        lr: &Range<utils::lsn::Lsn>,
+        is_image: bool,
+    ) -> Self {
+        LayerKey {
+            key: kr.start.to_i128()..kr.end.to_i128(),
+            lsn: lr.start.0..lr.end.0,
+            is_image,
+        }
+    }
+}
+
 /// Efficiently queryable layer coverage for each LSN.
 ///
 /// Allows answering layer map queries very efficiently,
@@ -417,6 +431,14 @@ impl<Value: Clone> BufferedHistoricLayerCoverage<Value> {
         }
     }
 
+    pub fn contains(&self, layer_key: &LayerKey) -> bool {
+        match self.buffer.get(layer_key) {
+            Some(None) => false,                         // layer remove was buffered
+            Some(_) => true,                             // layer insert was buffered
+            None => self.layers.contains_key(layer_key), // no buffered ops for this layer
+        }
+    }
+
     pub fn insert(&mut self, layer_key: LayerKey, value: Value) {
         self.buffer.insert(layer_key, Some(value));
     }

pageserver/src/tenant/storage_layer/delta_layer.rs

@@ -867,6 +867,14 @@ pub struct DeltaLayerWriter {
 }
 
 impl DeltaLayerWriter {
+    pub fn key_start(&self) -> Key {
+        self.inner.as_ref().unwrap().key_start
+    }
+
+    pub fn lsn_range(&self) -> Range<Lsn> {
+        self.inner.as_ref().unwrap().lsn_range.clone()
+    }
+
     ///
     /// Start building a new delta layer.
     ///

pageserver/src/tenant/timeline.rs

@@ -22,7 +22,8 @@ use tracing::*;
 use utils::id::TenantTimelineId;
 
 use std::cmp::{max, min, Ordering};
-use std::collections::{BinaryHeap, HashMap};
+use std::collections::BinaryHeap;
+use std::collections::HashMap;
 use std::fs;
 use std::ops::{Deref, Range};
 use std::path::{Path, PathBuf};
@@ -47,7 +48,7 @@ use crate::tenant::{
 };
 
 use crate::config::PageServerConf;
-use crate::keyspace::{KeyPartitioning, KeySpace, KeySpaceRandomAccum};
+use crate::keyspace::{KeyPartitioning, KeySpace};
 use crate::metrics::{TimelineMetrics, UNEXPECTED_ONDEMAND_DOWNLOADS};
 use crate::pgdatadir_mapping::LsnForTimestamp;
 use crate::pgdatadir_mapping::{is_rel_fsm_block_key, is_rel_vm_block_key};
@@ -122,17 +123,6 @@ pub struct Timeline {
 
     pub(super) layers: RwLock<LayerMap<dyn PersistentLayer>>,
 
-    /// Set of key ranges which should be covered by image layers to
-    /// allow GC to remove old layers. This set is created by GC and its cutoff LSN is also stored.
-    /// It is used by compaction task when it checks if new image layer should be created.
-    /// Newly created image layer doesn't help to remove the delta layer, until the
-    /// newly created image layer falls off the PITR horizon. So on next GC cycle,
-    /// gc_timeline may still want the new image layer to be created. To avoid redundant
-    /// image layers creation we should check if image layer exists but beyond PITR horizon.
-    /// This is why we need remember GC cutoff LSN.
-    ///
-    wanted_image_layers: Mutex<Option<(Lsn, KeySpace)>>,
-
     last_freeze_at: AtomicLsn,
     // Atomic would be more appropriate here.
     last_freeze_ts: RwLock<Instant>,
@@ -1364,7 +1354,6 @@ impl Timeline {
                 tenant_id,
                 pg_version,
                 layers: RwLock::new(LayerMap::default()),
-                wanted_image_layers: Mutex::new(None),
 
                 walredo_mgr,
                 walreceiver,
@@ -2915,30 +2904,6 @@ impl Timeline {
         let layers = self.layers.read().unwrap();
 
         let mut max_deltas = 0;
-        {
-            let wanted_image_layers = self.wanted_image_layers.lock().unwrap();
-            if let Some((cutoff_lsn, wanted)) = &*wanted_image_layers {
-                let img_range =
-                    partition.ranges.first().unwrap().start..partition.ranges.last().unwrap().end;
-                if wanted.overlaps(&img_range) {
-                    //
-                    // gc_timeline only pays attention to image layers that are older than the GC cutoff,
-                    // but create_image_layers creates image layers at last-record-lsn.
-                    // So it's possible that gc_timeline wants a new image layer to be created for a key range,
-                    // but the range is already covered by image layers at more recent LSNs. Before we
-                    // create a new image layer, check if the range is already covered at more recent LSNs.
-                    if !layers
-                        .image_layer_exists(&img_range, &(Lsn::min(lsn, *cutoff_lsn)..lsn + 1))?
-                    {
-                        debug!(
-                            "Force generation of layer {}-{} wanted by GC, cutoff={}, lsn={})",
-                            img_range.start, img_range.end, cutoff_lsn, lsn
-                        );
-                        return Ok(true);
-                    }
-                }
-            }
-        }
 
         for part_range in &partition.ranges {
             let image_coverage = layers.image_coverage(part_range, lsn)?;
@@ -3058,12 +3023,6 @@ impl Timeline {
                 image_layers.push(image_layer);
             }
         }
-        // All layers that the GC wanted us to create have now been created.
-        //
-        // It's possible that another GC cycle happened while we were compacting, and added
-        // something new to wanted_image_layers, and we now clear that before processing it.
-        // That's OK, because the next GC iteration will put it back in.
-        *self.wanted_image_layers.lock().unwrap() = None;
 
         // Sync the new layer to disk before adding it to the layer map, to make sure
         // we don't garbage collect something based on the new layer, before it has
@@ -3416,7 +3375,54 @@ impl Timeline {
                         || contains_hole
                     {
                         // ... if so, flush previous layer and prepare to write new one
-                        new_layers.push(writer.take().unwrap().finish(prev_key.unwrap().next())?);
+
+                        let end_key = prev_key.unwrap().next();
+
+                        let w = writer.take().unwrap();
+
+                        // If an identical L1 layer already exists, no need to create a new one.
+                        //
+                        // This can happen if compaction is interrupted after it has already
+                        // created some or all of the L1 layers, but has not deleted the L0 layers
+                        // yet, so that on next compaction, we do the same work again.
+                        //
+                        // NOTE: this is racy, if there can be any other task that concurrently
+                        // creates L1 layers. Currently, there can be only one compaction task
+                        // running at any time, so this is fine.
+                        //
+                        // Also we hold `layer_removal_cs` guard which should prevent race condition
+                        // even if there are two or more concurrent compaction tasks.
+                        //
+                        // But there is an opposite issue: we check presence of duplicates under
+                        // `layers` shared lock, but then it is released. So there is a gap between
+                        // this check and adding new layer to layer map. In principle in this gap some
+                        // some other task (i.e. GC) can drop this layer and we already abandon insertion
+                        // of duplicate layer. As a result there will be no such layer at all.
+                        // In other words: we have some state S1 of pageserver where layer L1 can be removed by GC.
+                        // Then we run compaction and it switch pageserver to the state S2 which writes duplicate of
+                        // layer L1 and where it can not be removed. With this patch it is possible that
+                        // we switch pageserver to state S2 but... with L1 lost.
+                        // It is just hypothetical situation and there is no such concrete scenario which
+                        // reproduces this problem. So let's take this risk.
+                        //
+                        if self.layers.read().unwrap().contains(
+                            &(w.key_start()..end_key),
+                            &w.lsn_range(),
+                            false, // not an image layer
+                        ) {
+                            info!(
+                                "Skip generation of duplicate layer {}_{}__{}_{}",
+                                w.key_start(),
+                                end_key,
+                                w.lsn_range().start,
+                                w.lsn_range().end
+                            );
+                            drop(w);
+                        } else {
+                            let new_layer = w.finish(end_key)?;
+
+                            new_layers.push(new_layer);
+                        }
                         writer = None;
 
                         if contains_hole {
@@ -3473,6 +3479,10 @@ impl Timeline {
 
         drop(all_keys_iter); // So that deltas_to_compact is no longer borrowed
 
+        fail_point!("compact-level0-phase1-finish", |_| {
+            Err(anyhow::anyhow!("failpoint compact-level0-phase1-finish").into())
+        });
+
         Ok(CompactLevel0Phase1Result {
             new_layers,
             deltas_to_compact,
@@ -3667,7 +3677,7 @@ impl Timeline {
     /// within a layer file. We can only remove the whole file if it's fully
     /// obsolete.
     ///
-    pub(super) async fn gc(&self, ctx: &RequestContext) -> anyhow::Result<GcResult> {
+    pub(super) async fn gc(&self) -> anyhow::Result<GcResult> {
         let timer = self.metrics.garbage_collect_histo.start_timer();
 
         fail_point!("before-timeline-gc");
@@ -3697,7 +3707,6 @@ impl Timeline {
                 pitr_cutoff,
                 retain_lsns,
                 new_gc_cutoff,
-                ctx,
             )
             .instrument(
                 info_span!("gc_timeline", timeline = %self.timeline_id, cutoff = %new_gc_cutoff),
@@ -3717,7 +3726,6 @@ impl Timeline {
         pitr_cutoff: Lsn,
         retain_lsns: Vec<Lsn>,
         new_gc_cutoff: Lsn,
-        ctx: &RequestContext,
     ) -> anyhow::Result<GcResult> {
         let now = SystemTime::now();
         let mut result: GcResult = GcResult::default();
@@ -3763,16 +3771,6 @@ impl Timeline {
         }
 
         let mut layers_to_remove = Vec::new();
-        let mut wanted_image_layers = KeySpaceRandomAccum::default();
-        // Do not collect keyspace for Unit tests
-        let gc_keyspace = if ctx.task_kind() == TaskKind::GarbageCollector {
-            Some(
-                self.collect_keyspace(self.get_last_record_lsn(), ctx)
-                    .await?,
-            )
-        } else {
-            None
-        };
 
         // Scan all layers in the timeline (remote or on-disk).
         //
@@ -3856,21 +3854,6 @@ impl Timeline {
                     "keeping {} because it is the latest layer",
                     l.filename().file_name()
                 );
-                // Collect delta key ranges that need image layers to allow garbage
-                // collecting the layers.
-                // It is not so obvious whether we need to propagate information only about
-                // 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 let Some(keyspace) = &gc_keyspace {
-                        let layer_logical_size = keyspace.get_logical_size(&l.get_key_range());
-                        let layer_age = new_gc_cutoff.0 - l.get_lsn_range().start.0;
-                        if layer_logical_size <= layer_age {
-                            wanted_image_layers.add_range(l.get_key_range());
-                        }
-                    }
-                }
                 result.layers_not_updated += 1;
                 continue 'outer;
             }
@@ -3883,10 +3866,6 @@ impl Timeline {
             );
             layers_to_remove.push(Arc::clone(&l));
         }
-        self.wanted_image_layers
-            .lock()
-            .unwrap()
-            .replace((new_gc_cutoff, wanted_image_layers.to_keyspace()));
 
         let mut updates = layers.batch_update();
         if !layers_to_remove.is_empty() {

test_runner/performance/test_gc_feedback.py

@@ -1,76 +0,0 @@
-import pytest
-from fixtures.benchmark_fixture import MetricReport, NeonBenchmarker
-from fixtures.log_helper import log
-from fixtures.neon_fixtures import NeonEnvBuilder
-
-
-@pytest.mark.timeout(10000)
-def test_gc_feedback(neon_env_builder: NeonEnvBuilder, zenbenchmark: NeonBenchmarker):
-    """
-    Test that GC is able to collect all old layers even if them are forming
-    "stairs" and there are not three delta layers since last image layer.
-
-    Information about image layers needed to collect old layers should
-    be propagated by GC to compaction task which should take in in account
-    when make a decision which new image layers needs to be created.
-    """
-    env = neon_env_builder.init_start()
-    client = env.pageserver.http_client()
-
-    tenant_id, _ = env.neon_cli.create_tenant(
-        conf={
-            # disable default GC and compaction
-            "gc_period": "1000 m",
-            "compaction_period": "0 s",
-            "gc_horizon": f"{1024 ** 2}",
-            "checkpoint_distance": f"{1024 ** 2}",
-            "compaction_target_size": f"{1024 ** 2}",
-            # set PITR interval to be small, so we can do GC
-            "pitr_interval": "10 s",
-            # "compaction_threshold": "3",
-            # "image_creation_threshold": "2",
-        }
-    )
-    endpoint = env.endpoints.create_start("main", tenant_id=tenant_id)
-    timeline_id = endpoint.safe_psql("show neon.timeline_id")[0][0]
-    n_steps = 10
-    n_update_iters = 100
-    step_size = 10000
-    with endpoint.cursor() as cur:
-        cur.execute("SET statement_timeout='1000s'")
-        cur.execute(
-            "CREATE TABLE t(step bigint, count bigint default 0, payload text default repeat(' ', 100))  with (fillfactor=50)"
-        )
-        cur.execute("CREATE INDEX ON t(step)")
-        # In each step, we insert 'step_size' new rows, and update the newly inserted rows
-        # 'n_update_iters' times. This creates a lot of churn and generates lots of WAL at the end of the table,
-        # without modifying the earlier parts of the table.
-        for step in range(n_steps):
-            cur.execute(f"INSERT INTO t (step) SELECT {step} FROM generate_series(1, {step_size})")
-            for i in range(n_update_iters):
-                cur.execute(f"UPDATE t set count=count+1 where step = {step}")
-                cur.execute("vacuum t")
-
-            # cur.execute("select pg_table_size('t')")
-            # logical_size = cur.fetchone()[0]
-            logical_size = client.timeline_detail(tenant_id, timeline_id)["current_logical_size"]
-            log.info(f"Logical storage size  {logical_size}")
-
-            client.timeline_checkpoint(tenant_id, timeline_id)
-
-            # Do compaction and GC
-            client.timeline_gc(tenant_id, timeline_id, 0)
-            client.timeline_compact(tenant_id, timeline_id)
-            # One more iteration to check that no excessive image layers are generated
-            client.timeline_gc(tenant_id, timeline_id, 0)
-            client.timeline_compact(tenant_id, timeline_id)
-
-            physical_size = client.timeline_detail(tenant_id, timeline_id)["current_physical_size"]
-            log.info(f"Physical storage size {physical_size}")
-
-    MB = 1024 * 1024
-    zenbenchmark.record("logical_size", logical_size // MB, "Mb", MetricReport.LOWER_IS_BETTER)
-    zenbenchmark.record("physical_size", physical_size // MB, "Mb", MetricReport.LOWER_IS_BETTER)
-    zenbenchmark.record(
-        "physical/logical ratio", physical_size / logical_size, "", MetricReport.LOWER_IS_BETTER
-    )

test_runner/regress/test_duplicate_layers.py

@@ -0,0 +1,42 @@
+import time
+
+import pytest
+from fixtures.neon_fixtures import NeonEnvBuilder, PgBin
+
+
+# Test duplicate layer detection
+#
+# This test sets fail point at the end of first compaction phase:
+# after flushing new L1 layers but before deletion of L0 layes
+# It should cause generation of duplicate L1 layer by compaction after restart
+@pytest.mark.timeout(600)
+def test_duplicate_layers(neon_env_builder: NeonEnvBuilder, pg_bin: PgBin):
+    env = neon_env_builder.init_start()
+
+    # These warnings are expected, when the pageserver is restarted abruptly
+    env.pageserver.allowed_errors.append(".*found future image layer.*")
+    env.pageserver.allowed_errors.append(".*found future delta layer.*")
+    env.pageserver.allowed_errors.append(".*duplicate layer.*")
+
+    pageserver_http = env.pageserver.http_client()
+
+    # Use aggressive compaction and checkpoint settings
+    tenant_id, _ = env.neon_cli.create_tenant(
+        conf={
+            "checkpoint_distance": f"{1024 ** 2}",
+            "compaction_target_size": f"{1024 ** 2}",
+            "compaction_period": "1 s",
+            "compaction_threshold": "3",
+        }
+    )
+    endpoint = env.endpoints.create_start("main", tenant_id=tenant_id)
+    connstr = endpoint.connstr(options="-csynchronous_commit=off")
+    pg_bin.run_capture(["pgbench", "-i", "-s10", connstr])
+
+    pageserver_http.configure_failpoints(("compact-level0-phase1-finish", "exit"))
+
+    with pytest.raises(Exception):
+        pg_bin.run_capture(["pgbench", "-P1", "-N", "-c5", "-T500", "-Mprepared", connstr])
+    env.pageserver.stop()
+    env.pageserver.start()
+    time.sleep(10)  # let compaction to be performed