latency recorder sketches

Conflicts:
	pageserver/src/tenant/timeline.rs
	test_runner/fixtures/neon_fixtures.py

.github/workflows/build_and_test.yml

@@ -892,14 +892,14 @@ jobs:
         run: |
           for repo in neondatabase 369495373322.dkr.ecr.eu-central-1.amazonaws.com; do
             docker buildx imagetools create -t $repo/neon:latest \
-                                               neondatabase/neon:${{ needs.tag.outputs.build-tag }}
+                                               $repo/neon:${{ needs.tag.outputs.build-tag }}
 
             for version in ${VERSIONS}; do
               docker buildx imagetools create -t $repo/compute-node-${version}:latest \
-                                                 neondatabase/compute-node-${version}:${{ needs.tag.outputs.build-tag }}
+                                                 $repo/compute-node-${version}:${{ needs.tag.outputs.build-tag }}
 
               docker buildx imagetools create -t $repo/vm-compute-node-${version}:latest \
-                                                 neondatabase/vm-compute-node-${version}:${{ needs.tag.outputs.build-tag }}
+                                                 $repo/vm-compute-node-${version}:${{ needs.tag.outputs.build-tag }}
             done
           done
           docker buildx imagetools create -t neondatabase/neon-test-extensions-v16:latest \

.github/workflows/release.yml

@@ -3,9 +3,8 @@ name: Create Release Branch
 on:
   schedule:
     # It should be kept in sync with if-condition in jobs
-    - cron: '0 6 * * THU' # Proxy release
     - cron: '0 6 * * FRI' # Storage release
-    - cron: '0 7 * * FRI' # Compute release
+    - cron: '0 6 * * THU' # Proxy release
   workflow_dispatch:
     inputs:
       create-storage-release-branch:
@@ -56,7 +55,7 @@ jobs:
       ci-access-token: ${{ secrets.CI_ACCESS_TOKEN }}
 
   create-compute-release-branch:
-    if: ${{ github.event.schedule == '0 7 * * FRI' || inputs.create-compute-release-branch }}
+    if: inputs.create-compute-release-branch
 
     permissions:
       contents: write

compute/compute-node.Dockerfile

@@ -67,9 +67,6 @@ RUN cd postgres && \
     # Enable some of contrib extensions
     echo 'trusted = true' >> /usr/local/pgsql/share/extension/autoinc.control && \
     echo 'trusted = true' >> /usr/local/pgsql/share/extension/dblink.control && \
-    echo 'trusted = true' >> /usr/local/pgsql/share/extension/postgres_fdw.control && \
-    file=/usr/local/pgsql/share/extension/postgres_fdw--1.0.sql && [ -e $file ] && \
-    echo 'GRANT USAGE ON FOREIGN DATA WRAPPER postgres_fdw TO neon_superuser;' >> $file && \
     echo 'trusted = true' >> /usr/local/pgsql/share/extension/bloom.control && \
     echo 'trusted = true' >> /usr/local/pgsql/share/extension/earthdistance.control && \
     echo 'trusted = true' >> /usr/local/pgsql/share/extension/insert_username.control && \

compute_tools/src/compute.rs

@@ -41,14 +41,14 @@ use crate::local_proxy;
 use crate::pg_helpers::*;
 use crate::spec::*;
 use crate::spec_apply::ApplySpecPhase::{
-    CreateAndAlterDatabases, CreateAndAlterRoles, CreateAvailabilityCheck, CreateSchemaNeon,
-    CreateSuperUser, DropInvalidDatabases, DropRoles, FinalizeDropLogicalSubscriptions,
-    HandleNeonExtension, HandleOtherExtensions, RenameAndDeleteDatabases, RenameRoles,
-    RunInEachDatabase,
+    CreateAndAlterDatabases, CreateAndAlterRoles, CreateAvailabilityCheck, CreateSuperUser,
+    DropInvalidDatabases, DropRoles, HandleNeonExtension, HandleOtherExtensions,
+    RenameAndDeleteDatabases, RenameRoles, RunInEachDatabase,
 };
 use crate::spec_apply::PerDatabasePhase;
 use crate::spec_apply::PerDatabasePhase::{
-    ChangeSchemaPerms, DeleteDBRoleReferences, DropLogicalSubscriptions, HandleAnonExtension,
+    ChangeSchemaPerms, DeleteDBRoleReferences, DropSubscriptionsForDeletedDatabases,
+    HandleAnonExtension,
 };
 use crate::spec_apply::{apply_operations, MutableApplyContext, DB};
 use crate::sync_sk::{check_if_synced, ping_safekeeper};
@@ -340,15 +340,6 @@ impl ComputeNode {
         self.state.lock().unwrap().status
     }
 
-    pub fn get_timeline_id(&self) -> Option<TimelineId> {
-        self.state
-            .lock()
-            .unwrap()
-            .pspec
-            .as_ref()
-            .map(|s| s.timeline_id)
-    }
-
     // Remove `pgdata` directory and create it again with right permissions.
     fn create_pgdata(&self) -> Result<()> {
         // Ignore removal error, likely it is a 'No such file or directory (os error 2)'.
@@ -938,48 +929,6 @@ impl ComputeNode {
                 .map(|role| (role.name.clone(), role))
                 .collect::<HashMap<String, Role>>();
 
-            // Check if we need to drop subscriptions before starting the endpoint.
-            //
-            // It is important to do this operation exactly once when endpoint starts on a new branch.
-            // Otherwise, we may drop not inherited, but newly created subscriptions.
-            //
-            // We cannot rely only on spec.drop_subscriptions_before_start flag,
-            // because if for some reason compute restarts inside VM,
-            // it will start again with the same spec and flag value.
-            //
-            // To handle this, we save the fact of the operation in the database
-            // in the neon.drop_subscriptions_done table.
-            // If the table does not exist, we assume that the operation was never performed, so we must do it.
-            // If table exists, we check if the operation was performed on the current timelilne.
-            //
-            let mut drop_subscriptions_done = false;
-
-            if spec.drop_subscriptions_before_start {
-                let timeline_id = self.get_timeline_id().context("timeline_id must be set")?;
-                let query = format!("select 1 from neon.drop_subscriptions_done where timeline_id = '{}'", timeline_id);
-
-                info!("Checking if drop subscription operation was already performed for timeline_id: {}", timeline_id);
-
-                drop_subscriptions_done =  match
-                    client.simple_query(&query).await {
-                    Ok(result) => {
-                        matches!(&result[0], postgres::SimpleQueryMessage::Row(_))
-                    },
-                    Err(e) =>
-                    {
-                        match e.code() {
-                            Some(&SqlState::UNDEFINED_TABLE) => false,
-                            _ => {
-                                // We don't expect any other error here, except for the schema/table not existing
-                                error!("Error checking if drop subscription operation was already performed: {}", e);
-                                return Err(e.into());
-                            }
-                        }
-                    }
-                }
-            };
-
-
             let jwks_roles = Arc::new(
                 spec.as_ref()
                     .local_proxy_config
@@ -1047,7 +996,7 @@ impl ComputeNode {
                         jwks_roles.clone(),
                         concurrency_token.clone(),
                         db,
-                        [DropLogicalSubscriptions].to_vec(),
+                        [DropSubscriptionsForDeletedDatabases].to_vec(),
                     );
 
                     Ok(spawn(fut))
@@ -1075,7 +1024,6 @@ impl ComputeNode {
                 CreateAndAlterRoles,
                 RenameAndDeleteDatabases,
                 CreateAndAlterDatabases,
-                CreateSchemaNeon,
             ] {
                 info!("Applying phase {:?}", &phase);
                 apply_operations(
@@ -1116,17 +1064,6 @@ impl ComputeNode {
                     }
 
                     let conf = Arc::new(conf);
-                    let mut phases = vec![
-                        DeleteDBRoleReferences,
-                        ChangeSchemaPerms,
-                        HandleAnonExtension,
-                    ];
-
-                    if spec.drop_subscriptions_before_start && !drop_subscriptions_done {
-                        info!("Adding DropLogicalSubscriptions phase because drop_subscriptions_before_start is set");
-                        phases.push(DropLogicalSubscriptions);
-                    }
-
                     let fut = Self::apply_spec_sql_db(
                         spec.clone(),
                         conf,
@@ -1134,7 +1071,12 @@ impl ComputeNode {
                         jwks_roles.clone(),
                         concurrency_token.clone(),
                         db,
-                        phases,
+                        [
+                            DeleteDBRoleReferences,
+                            ChangeSchemaPerms,
+                            HandleAnonExtension,
+                        ]
+                        .to_vec(),
                     );
 
                     Ok(spawn(fut))
@@ -1146,20 +1088,12 @@ impl ComputeNode {
                 handle.await??;
             }
 
-            let mut phases = vec![
+            for phase in vec![
                 HandleOtherExtensions,
-                HandleNeonExtension, // This step depends on CreateSchemaNeon
+                HandleNeonExtension,
                 CreateAvailabilityCheck,
                 DropRoles,
-            ];
-
-            // This step depends on CreateSchemaNeon
-            if spec.drop_subscriptions_before_start && !drop_subscriptions_done {
-                info!("Adding FinalizeDropLogicalSubscriptions phase because drop_subscriptions_before_start is set");
-                phases.push(FinalizeDropLogicalSubscriptions);
-            }
-
-            for phase in phases {
+            ] {
                 debug!("Applying phase {:?}", &phase);
                 apply_operations(
                     spec.clone(),
@@ -1529,14 +1463,6 @@ impl ComputeNode {
                         Ok(())
                     },
                 )?;
-
-                let postgresql_conf_path = pgdata_path.join("postgresql.conf");
-                if config::line_in_file(
-                    &postgresql_conf_path,
-                    "neon.disable_logical_replication_subscribers=false",
-                )? {
-                    info!("updated postgresql.conf to set neon.disable_logical_replication_subscribers=false");
-                }
                 self.pg_reload_conf()?;
             }
             self.post_apply_config()?;

compute_tools/src/config.rs

@@ -129,13 +129,6 @@ pub fn write_postgres_conf(
 
     writeln!(file, "neon.extension_server_port={}", extension_server_port)?;
 
-    if spec.drop_subscriptions_before_start {
-        writeln!(file, "neon.disable_logical_replication_subscribers=true")?;
-    } else {
-        // be explicit about the default value
-        writeln!(file, "neon.disable_logical_replication_subscribers=false")?;
-    }
-
     // This is essential to keep this line at the end of the file,
     // because it is intended to override any settings above.
     writeln!(file, "include_if_exists = 'compute_ctl_temp_override.conf'")?;

compute_tools/src/spec_apply.rs

@@ -47,7 +47,7 @@ pub enum PerDatabasePhase {
     DeleteDBRoleReferences,
     ChangeSchemaPerms,
     HandleAnonExtension,
-    DropLogicalSubscriptions,
+    DropSubscriptionsForDeletedDatabases,
 }
 
 #[derive(Clone, Debug)]
@@ -58,13 +58,11 @@ pub enum ApplySpecPhase {
     CreateAndAlterRoles,
     RenameAndDeleteDatabases,
     CreateAndAlterDatabases,
-    CreateSchemaNeon,
     RunInEachDatabase { db: DB, subphase: PerDatabasePhase },
     HandleOtherExtensions,
     HandleNeonExtension,
     CreateAvailabilityCheck,
     DropRoles,
-    FinalizeDropLogicalSubscriptions,
 }
 
 pub struct Operation {
@@ -333,7 +331,7 @@ async fn get_operations<'a>(
                             // NB: there could be other db states, which prevent us from dropping
                             // the database. For example, if db is used by any active subscription
                             // or replication slot.
-                            // Such cases are handled in the DropLogicalSubscriptions
+                            // Such cases are handled in the DropSubscriptionsForDeletedDatabases
                             // phase. We do all the cleanup before actually dropping the database.
                             let drop_db_query: String = format!(
                                 "DROP DATABASE IF EXISTS {} WITH (FORCE)",
@@ -444,19 +442,13 @@ async fn get_operations<'a>(
 
             Ok(Box::new(operations))
         }
-        ApplySpecPhase::CreateSchemaNeon => Ok(Box::new(once(Operation {
-            query: String::from("CREATE SCHEMA IF NOT EXISTS neon"),
-            comment: Some(String::from(
-                "create schema for neon extension and utils tables",
-            )),
-        }))),
         ApplySpecPhase::RunInEachDatabase { db, subphase } => {
             match subphase {
-                PerDatabasePhase::DropLogicalSubscriptions => {
+                PerDatabasePhase::DropSubscriptionsForDeletedDatabases => {
                     match &db {
                         DB::UserDB(db) => {
                             let drop_subscription_query: String = format!(
-                                include_str!("sql/drop_subscriptions.sql"),
+                                include_str!("sql/drop_subscription_for_drop_dbs.sql"),
                                 datname_str = escape_literal(&db.name),
                             );
 
@@ -674,6 +666,10 @@ async fn get_operations<'a>(
         }
         ApplySpecPhase::HandleNeonExtension => {
             let operations = vec![
+                Operation {
+                    query: String::from("CREATE SCHEMA IF NOT EXISTS neon"),
+                    comment: Some(String::from("init: add schema for extension")),
+                },
                 Operation {
                     query: String::from("CREATE EXTENSION IF NOT EXISTS neon WITH SCHEMA neon"),
                     comment: Some(String::from(
@@ -716,9 +712,5 @@ async fn get_operations<'a>(
 
             Ok(Box::new(operations))
         }
-        ApplySpecPhase::FinalizeDropLogicalSubscriptions => Ok(Box::new(once(Operation {
-            query: String::from(include_str!("sql/finalize_drop_subscriptions.sql")),
-            comment: None,
-        }))),
     }
 }

compute_tools/src/sql/finalize_drop_subscriptions.sql

@@ -1,21 +0,0 @@
-DO $$
-BEGIN
-    IF NOT EXISTS(
-        SELECT 1
-        FROM pg_catalog.pg_tables
-        WHERE tablename = 'drop_subscriptions_done'
-        AND schemaname = 'neon'
-    )
-    THEN
-        CREATE TABLE neon.drop_subscriptions_done
-        (id serial primary key, timeline_id text);
-    END IF;
-
-    -- preserve the timeline_id of the last drop_subscriptions run
-    -- to ensure that the cleanup of a timeline is executed only once.
-    -- use upsert to avoid the table bloat in case of cascade branching (branch of a branch)
-    INSERT INTO neon.drop_subscriptions_done VALUES (1, current_setting('neon.timeline_id'))
-    ON CONFLICT (id) DO UPDATE
-    SET timeline_id = current_setting('neon.timeline_id');
-END
-$$

control_plane/src/bin/neon_local.rs

@@ -1357,7 +1357,6 @@ async fn handle_endpoint(subcmd: &EndpointCmd, env: &local_env::LocalEnv) -> Res
                 args.pg_version,
                 mode,
                 !args.update_catalog,
-                false,
             )?;
         }
         EndpointCmd::Start(args) => {

control_plane/src/endpoint.rs

@@ -76,7 +76,6 @@ pub struct EndpointConf {
     http_port: u16,
     pg_version: u32,
     skip_pg_catalog_updates: bool,
-    drop_subscriptions_before_start: bool,
     features: Vec<ComputeFeature>,
 }
 
@@ -144,7 +143,6 @@ impl ComputeControlPlane {
         pg_version: u32,
         mode: ComputeMode,
         skip_pg_catalog_updates: bool,
-        drop_subscriptions_before_start: bool,
     ) -> Result<Arc<Endpoint>> {
         let pg_port = pg_port.unwrap_or_else(|| self.get_port());
         let http_port = http_port.unwrap_or_else(|| self.get_port() + 1);
@@ -164,7 +162,6 @@ impl ComputeControlPlane {
             // with this we basically test a case of waking up an idle compute, where
             // we also skip catalog updates in the cloud.
             skip_pg_catalog_updates,
-            drop_subscriptions_before_start,
             features: vec![],
         });
 
@@ -180,7 +177,6 @@ impl ComputeControlPlane {
                 pg_port,
                 pg_version,
                 skip_pg_catalog_updates,
-                drop_subscriptions_before_start,
                 features: vec![],
             })?,
         )?;
@@ -244,7 +240,6 @@ pub struct Endpoint {
     // Optimizations
     skip_pg_catalog_updates: bool,
 
-    drop_subscriptions_before_start: bool,
     // Feature flags
     features: Vec<ComputeFeature>,
 }
@@ -296,7 +291,6 @@ impl Endpoint {
             tenant_id: conf.tenant_id,
             pg_version: conf.pg_version,
             skip_pg_catalog_updates: conf.skip_pg_catalog_updates,
-            drop_subscriptions_before_start: conf.drop_subscriptions_before_start,
             features: conf.features,
         })
     }
@@ -631,7 +625,6 @@ impl Endpoint {
             shard_stripe_size: Some(shard_stripe_size),
             local_proxy_config: None,
             reconfigure_concurrency: 1,
-            drop_subscriptions_before_start: self.drop_subscriptions_before_start,
         };
         let spec_path = self.endpoint_path().join("spec.json");
         std::fs::write(spec_path, serde_json::to_string_pretty(&spec)?)?;

docker-compose/docker-compose.yml

@@ -185,8 +185,6 @@ services:
   neon-test-extensions:
     profiles: ["test-extensions"]
     image: ${REPOSITORY:-neondatabase}/neon-test-extensions-v${PG_TEST_VERSION:-16}:${TAG:-latest}
-    environment:
-      - PGPASSWORD=cloud_admin
     entrypoint:
       - "/bin/bash"
       - "-c"

docker-compose/run-tests.sh

@@ -7,10 +7,7 @@ LIST=$( (echo -e "${SKIP//","/"\n"}"; ls -d -- *-src) | sort | uniq -u)
 for d in ${LIST}
 do
        [ -d "${d}" ] || continue
-       if ! psql -w -c "select 1" >/dev/null; then
-          FAILED="${d} ${FAILED}"
-          break
-       fi
+    psql -c "select 1" >/dev/null || break
        USE_PGXS=1 make -C "${d}" installcheck || FAILED="${d} ${FAILED}"
 done
 [ -z "${FAILED}" ] && exit 0

if we just want to draw a pie chart.rs

@@ -0,0 +1,35 @@
+// If we just want to draw a pie chart of where latency goes,
+// then we can project the `if we want to draw a jaeger trace.rs` into these structures.
+// - fold `Vec<...>` into a sum
+// - fold `start_time` and `end_time` into deltas
+
+struct SmgrLatencyRecorder {
+    parse_request: u64,
+    downstairs: Arc<Mutex<Downstairs>>,
+    flush_response: u64,
+}
+
+struct Downstairs {
+    wait_for_execution: u64, // batching happens here
+    execution: Arc<Mutex<Execution>>,
+}
+
+struct Execution {
+    traverse_and_submit: Plan,
+    wait_for_io_completions_and_walredo: WaitForIoCompletionsAndWalredo,
+}
+
+struct Plan {
+    visit_layer: VisitLayer,
+    // implict remainder: time in the fringe code and traversing timeline ancestor graph
+}
+
+struct VisitLayer {
+    index_lookup: u64,
+    submit_io: u64,
+}
+
+struct WaitForIoCompletionsAndWalredo {
+    wait_for_io_completions: u64,
+    walredo: u64,
+}

if we want to draw a jaeger trace.rs

@@ -0,0 +1,69 @@
+// If we want to draw a Jaeger trace
+
+struct SmgrLatencyRecorder {
+    start_time: Instant,
+    parse_request: ParseRequest,
+    downstairs: Arc<Mutex<Downstairs>>,
+    flush_response: FlushResponse,
+    end_time: Instant,
+}
+
+struct ParseRequest {
+    start_time: Instant,
+    end_time: Instant,
+}
+
+struct FlushResponse {
+    start_time: Instant,
+    end_time: Instant,
+}
+
+struct Downstairs {
+    start_time: Instant,
+    wait_for_execution: WaitForExecution, // batching happens here
+    execution: Arc<Mutex<Execution>>,
+    end_time: Instant,
+}
+
+enum WaitForExecution {
+    start_time: Instant,
+    end_time: Instant,
+}
+
+struct Execution {
+    start_time: Instant,
+    traverse_and_submit: Plan,
+    wait_for_io_completions_and_walredo: Vec<Arc<Mutex<WaitForIoCompletionsAndWalredo>>>,
+    end_time: Instant,
+}
+
+struct Plan {
+    start_time: Instant,
+    visit_layer: Vec<VisitLayer>,
+    // implict remainder: time in the fringe code and traversing timeline ancestor graph
+    end_time: Instant,
+}
+
+struct VisitLayer {
+    start_time: Instant,
+    index_lookup: IndexLookup,
+    submit_io: SubmitIo,
+    end_time: Instant,
+}
+
+struct IndexLookup {
+    start_time: Instant,
+    end_time: Instant,
+}
+
+struct SubmitIo {
+    start_time: Instant,
+    end_time: Instant,
+}
+
+struct WaitForIoCompletionsAndWalredo {
+    start_time: Instant,
+    wait_for_io_completions: u64,
+    walredo: u64,
+    end_time: Instant,
+}

libs/compute_api/src/spec.rs

@@ -138,13 +138,6 @@ pub struct ComputeSpec {
     /// enough spare connections for reconfiguration process to succeed.
     #[serde(default = "default_reconfigure_concurrency")]
     pub reconfigure_concurrency: usize,
-
-    /// If set to true, the compute_ctl will drop all subscriptions before starting the
-    /// compute. This is needed when we start an endpoint on a branch, so that child
-    /// would not compete with parent branch subscriptions
-    /// over the same replication content from publisher.
-    #[serde(default)] // Default false
-    pub drop_subscriptions_before_start: bool,
 }
 
 /// Feature flag to signal `compute_ctl` to enable certain experimental functionality.

pageserver/src/context.rs

@@ -92,203 +92,94 @@
 use crate::task_mgr::TaskKind;
 
 // The main structure of this module, see module-level comment.
-#[derive(Debug)]
+#[derive(Debug, Default)]
 pub struct RequestContext {
-    task_kind: TaskKind,
-    download_behavior: DownloadBehavior,
-    access_stats_behavior: AccessStatsBehavior,
-    page_content_kind: PageContentKind,
+    latency_recording: Option<latency_recording::LatencyRecording>,
+    io_concurrency: Option<io_concurrency_propagation::IoConcurrencyPropagation>,
+    cancel: Option<cancellation::Cancellation>,
 }
 
-/// The kind of access to the page cache.
-#[derive(Clone, Copy, PartialEq, Eq, Debug, enum_map::Enum, strum_macros::IntoStaticStr)]
-pub enum PageContentKind {
-    Unknown,
-    DeltaLayerSummary,
-    DeltaLayerBtreeNode,
-    DeltaLayerValue,
-    ImageLayerSummary,
-    ImageLayerBtreeNode,
-    ImageLayerValue,
-    InMemoryLayer,
-}
-
-/// Desired behavior if the operation requires an on-demand download
-/// to proceed.
-#[derive(Clone, Copy, PartialEq, Eq, Debug)]
-pub enum DownloadBehavior {
-    /// Download the layer file. It can take a while.
-    Download,
-
-    /// Download the layer file, but print a warning to the log. This should be used
-    /// in code where the layer file is expected to already exist locally.
-    Warn,
-
-    /// Return a PageReconstructError::NeedsDownload error
-    Error,
-}
-
-/// Whether this request should update access times used in LRU eviction
-#[derive(Clone, Copy, PartialEq, Eq, Debug)]
-pub(crate) enum AccessStatsBehavior {
-    /// Update access times: this request's access to data should be taken
-    /// as a hint that the accessed layer is likely to be accessed again
-    Update,
-
-    /// Do not update access times: this request is accessing the layer
-    /// but does not want to indicate that the layer should be retained in cache,
-    /// perhaps because the requestor is a compaction routine that will soon cover
-    /// this layer with another.
-    Skip,
-}
-
-pub struct RequestContextBuilder {
-    inner: RequestContext,
-}
-
-impl RequestContextBuilder {
-    /// A new builder with default settings
-    pub fn new(task_kind: TaskKind) -> Self {
-        Self {
-            inner: RequestContext {
-                task_kind,
-                download_behavior: DownloadBehavior::Download,
-                access_stats_behavior: AccessStatsBehavior::Update,
-                page_content_kind: PageContentKind::Unknown,
-            },
-        }
-    }
-
-    pub fn extend(original: &RequestContext) -> Self {
-        Self {
-            // This is like a Copy, but avoid implementing Copy because ordinary users of
-            // RequestContext should always move or ref it.
-            inner: RequestContext {
-                task_kind: original.task_kind,
-                download_behavior: original.download_behavior,
-                access_stats_behavior: original.access_stats_behavior,
-                page_content_kind: original.page_content_kind,
-            },
-        }
-    }
-
-    /// Configure the DownloadBehavior of the context: whether to
-    /// download missing layers, and/or warn on the download.
-    pub fn download_behavior(mut self, b: DownloadBehavior) -> Self {
-        self.inner.download_behavior = b;
-        self
-    }
-
-    /// Configure the AccessStatsBehavior of the context: whether layer
-    /// accesses should update the access time of the layer.
-    pub(crate) fn access_stats_behavior(mut self, b: AccessStatsBehavior) -> Self {
-        self.inner.access_stats_behavior = b;
-        self
-    }
-
-    pub(crate) fn page_content_kind(mut self, k: PageContentKind) -> Self {
-        self.inner.page_content_kind = k;
-        self
-    }
-
-    pub fn build(self) -> RequestContext {
-        self.inner
-    }
+trait Propagatable: Default {
+    fn propagate(&self, child: &mut Self);
 }
 
 impl RequestContext {
-    /// Create a new RequestContext that has no parent.
-    ///
-    /// The function is called `new` because, once we add children
-    /// to it using `detached_child` or `attached_child`, the context
-    /// form a tree (not implemented yet since cancellation will be
-    /// the first feature that requires a tree).
-    ///
-    /// # Future: Cancellation
-    ///
-    /// The only reason why a context like this one can be canceled is
-    /// because someone explicitly canceled it.
-    /// It has no parent, so it cannot inherit cancellation from there.
-    pub fn new(task_kind: TaskKind, download_behavior: DownloadBehavior) -> Self {
-        RequestContextBuilder::new(task_kind)
-            .download_behavior(download_behavior)
-            .build()
+    fn root() -> Self {
+        Self::default()
     }
-
-    /// Create a detached child context for a task that may outlive `self`.
-    ///
-    /// Use this when spawning new background activity that should complete
-    /// even if the current request is canceled.
-    ///
-    /// # Future: Cancellation
-    ///
-    /// Cancellation of `self` will not propagate to the child context returned
-    /// by this method.
-    ///
-    /// # Future: Structured Concurrency
-    ///
-    /// We could add the Future as a parameter to this function, spawn it as a task,
-    /// and pass to the new task the child context as an argument.
-    /// That would be an ergonomic improvement.
-    ///
-    /// We could make new calls to this function fail if `self` is already canceled.
-    pub fn detached_child(&self, task_kind: TaskKind, download_behavior: DownloadBehavior) -> Self {
-        self.child_impl(task_kind, download_behavior)
-    }
-
-    /// Create a child of context `self` for a task that shall not outlive `self`.
-    ///
-    /// Use this when fanning-out work to other async tasks.
-    ///
-    /// # Future: Cancellation
-    ///
-    /// Cancelling a context will propagate to its attached children.
-    ///
-    /// # Future: Structured Concurrency
-    ///
-    /// We could add the Future as a parameter to this function, spawn it as a task,
-    /// and track its `JoinHandle` inside the `RequestContext`.
-    ///
-    /// We could then provide another method to allow waiting for all child tasks
-    /// to finish.
-    ///
-    /// We could make new calls to this function fail if `self` is already canceled.
-    /// Alternatively, we could allow the creation but not spawn the task.
-    /// The method to wait for child tasks would return an error, indicating
-    /// that the child task was not started because the context was canceled.
-    pub fn attached_child(&self) -> Self {
-        self.child_impl(self.task_kind(), self.download_behavior())
-    }
-
-    /// Use this function when you should be creating a child context using
-    /// [`attached_child`] or [`detached_child`], but your caller doesn't provide
-    /// a context and you are unwilling to change all callers to provide one.
-    ///
-    /// Before we add cancellation, we should get rid of this method.
-    ///
-    /// [`attached_child`]: Self::attached_child
-    /// [`detached_child`]: Self::detached_child
-    pub fn todo_child(task_kind: TaskKind, download_behavior: DownloadBehavior) -> Self {
-        Self::new(task_kind, download_behavior)
-    }
-
-    fn child_impl(&self, task_kind: TaskKind, download_behavior: DownloadBehavior) -> Self {
-        Self::new(task_kind, download_behavior)
-    }
-
-    pub fn task_kind(&self) -> TaskKind {
-        self.task_kind
-    }
-
-    pub fn download_behavior(&self) -> DownloadBehavior {
-        self.download_behavior
-    }
-
-    pub(crate) fn access_stats_behavior(&self) -> AccessStatsBehavior {
-        self.access_stats_behavior
-    }
-
-    pub(crate) fn page_content_kind(&self) -> PageContentKind {
-        self.page_content_kind
+    fn child(&self) -> RequestContext {
+        let mut child = RequestContext::default();
+        let Self {
+            latency_recording,
+        } = self;
+        if let Some(latency_recording) = latency_recording {
+            child.latency_recording = Some(latency_recording.child());
+        }
     }
 }
+
+mod latency_recording {
+    struct LatencyRecording {
+        inner: Arc<Mutex<Inner>>,
+    }
+
+    impl LatencyRecording {
+        fn new() -> Self {
+            Self {
+                current: Mutex::new(HashMap::new()),
+            }
+        }
+
+        fn on_request_recv(&self, now: Instant) -> {
+            let mut inner = self.inner.lock().unwrap();
+            inner.current.insert(now, now);
+        }
+
+    }
+
+    impl Propagatable for LatencyRecording {
+        fn propagate(&self, other: &mut Self) {
+            let mut inner = self.inner.lock().unwrap();
+            let other_inner = other.get_mut();
+            for (k, v) in other_inner.current.iter() {
+                inner.current.insert(*k, *v);
+            }
+        }
+    }
+}
+
+mod io_concurrency_propagation {
+
+    struct IoConcurrencyPropagation {
+        inner: IoConcurrency,
+    }
+
+    impl Propagatable for IoConcurrencyPropagation {
+        fn propagate(&self, other: &mut Self) {
+            other.inner = self.inner.clone();
+        }
+    }
+
+}
+
+mod cancellation {
+
+    struct Cancellation {
+        sources: Vec<CancellationToken>,
+    }
+
+    impl Propagatable for Cancellation {
+        fn propagate(&self, other: &mut Self) {
+            other.sources.extend(self.sources.iter().map(|tok| tok.child_toke()));
+        }
+    }
+
+    impl Cancellation {
+        async fn cancelled(&self) {
+            // TODO: this one is quite inefficient, it allocates
+            // But it's clear something better can be built within this architecture.
+            futures::future::select_all(self.sources.iter().map(|tok| tok.cancelled()))
+        }
+    }
+
+}

pageserver/src/page_service.rs

@@ -728,6 +728,12 @@ impl PageServerHandler {
             msg = pgb.read_message() => { msg }
         };
 
+        let ctx = ctx.enrich(
+            SmgrOpLatencyRecorder {
+                start_time: Instant::now(),
+                parse_request: ParseRequest::unset(),
+            });
+
         let received_at = Instant::now();
 
         let copy_data_bytes = match msg? {
@@ -748,9 +754,15 @@ impl PageServerHandler {
 
         fail::fail_point!("ps::handle-pagerequest-message");
 
+
+        {
+
+            let parse_request_recorder= ctx.get::<SmgrOpLatencyRecorder>().parse_request;
         // parse request
         let neon_fe_msg =
-            PagestreamFeMessage::parse(&mut copy_data_bytes.reader(), protocol_version)?;
+            PagestreamFeMessage::parse(&mut copy_data_bytes.reader(), protocol_version, &parse_request_recorder)?;
+        }
+
 
         // TODO: turn in to async closure once available to avoid repeating received_at
         async fn record_op_start_and_throttle(
@@ -918,6 +930,11 @@ impl PageServerHandler {
                     shard: shard.downgrade(),
                     effective_request_lsn,
                     pages: smallvec::smallvec![BatchedGetPageRequest { req, timer }],
+                    downstairs: {
+                        let downstairs = Arc::new(Mutex::new(Downstairs::init()));
+                        *(&mut ctx.get::<SmgrOpLatencyRecorder>().downstairs) = Arc::clone(&downstairs);
+                        downstairs
+                    },
                 }
             }
             #[cfg(feature = "testing")]
@@ -963,12 +980,14 @@ impl PageServerHandler {
                     shard: accum_shard,
                     pages: ref mut accum_pages,
                     effective_request_lsn: accum_lsn,
+                    recorder: accum_recorder,
                 }),
                 BatchedFeMessage::GetPage {
                     span: _,
                     shard: this_shard,
                     pages: this_pages,
                     effective_request_lsn: this_lsn,
+                    recorder: this_recorder,
                 },
             ) if (|| {
                 assert_eq!(this_pages.len(), 1);
@@ -994,6 +1013,7 @@ impl PageServerHandler {
             {
                 // ok to batch
                 accum_pages.extend(this_pages);
+                *this_recorder = accum_recorder;
                 Ok(())
             }
             #[cfg(feature = "testing")]
@@ -1589,6 +1609,10 @@ impl PageServerHandler {
             async move {
                 let _cancel_batcher = cancel_batcher.drop_guard();
                 loop {
+                    let ctx = ctx.attached_child();
+
+                    let child = ctx.attached_child();
+
                     let maybe_batch = batch_rx.recv().await;
                     let batch = match maybe_batch {
                         Ok(batch) => batch,
@@ -1603,6 +1627,8 @@ impl PageServerHandler {
                             return Err(e);
                         }
                     };
+                    batch.downstairs.wait_for_execution.end_time = Instant::now();
+                    ctx.latency_recording.enrich(batch.downstairs);
                     self.pagesteam_handle_batched_message(
                         pgb_writer,
                         batch,

pageserver/src/tenant/timeline/compaction.rs

@@ -436,14 +436,12 @@ impl KeyHistoryRetention {
         if dry_run {
             return true;
         }
-        let layer_generation;
-        {
-            let guard = tline.layers.read().await;
-            if !guard.contains_key(key) {
-                return false;
-            }
-            layer_generation = guard.get_from_key(key).metadata().generation;
+        let guard = tline.layers.read().await;
+        if !guard.contains_key(key) {
+            return false;
         }
+        let layer_generation = guard.get_from_key(key).metadata().generation;
+        drop(guard);
         if layer_generation == tline.generation {
             info!(
                 key=%key,
@@ -2140,11 +2138,6 @@ impl Timeline {
             self.get_gc_compaction_watermark()
         };
 
-        if compact_below_lsn == Lsn::INVALID {
-            tracing::warn!("no layers to compact with gc: gc_cutoff not generated yet, skipping gc bottom-most compaction");
-            return Ok(vec![]);
-        }
-
         // Split compaction job to about 4GB each
         const GC_COMPACT_MAX_SIZE_MB: u64 = 4 * 1024;
         let sub_compaction_max_job_size_mb =
@@ -2219,12 +2212,6 @@ impl Timeline {
                 } else {
                     end
                 };
-                let end = if ranges_num == idx + 1 {
-                    // extend the compaction range to the end of the key range if it's the last partition
-                    end.max(job.compact_key_range.end)
-                } else {
-                    end
-                };
                 info!(
                     "splitting compaction job: {}..{}, estimated_size={}",
                     start, end, total_size
@@ -2345,11 +2332,6 @@ impl Timeline {
                 // each of the retain_lsn. Therefore, if the user-provided `compact_lsn_range.end` is larger than the real gc cutoff, we will use
                 // the real cutoff.
                 let mut gc_cutoff = if compact_lsn_range.end == Lsn::MAX {
-                    if real_gc_cutoff == Lsn::INVALID {
-                        // If the gc_cutoff is not generated yet, we should not compact anything.
-                        tracing::warn!("no layers to compact with gc: gc_cutoff not generated yet, skipping gc bottom-most compaction");
-                        return Ok(());
-                    }
                     real_gc_cutoff
                 } else {
                     compact_lsn_range.end
@@ -2881,7 +2863,7 @@ impl Timeline {
             "produced {} delta layers and {} image layers, {} layers are kept",
             produced_delta_layers_len,
             produced_image_layers_len,
-            keep_layers.len()
+            layer_selection.len()
         );
 
         // Step 3: Place back to the layer map.
@@ -2927,28 +2909,8 @@ impl Timeline {
         // be batched into `schedule_compaction_update`.
         let disk_consistent_lsn = self.disk_consistent_lsn.load();
         self.schedule_uploads(disk_consistent_lsn, None)?;
-        // If a layer gets rewritten throughout gc-compaction, we need to keep that layer only in `compact_to` instead
-        // of `compact_from`.
-        let compact_from = {
-            let mut compact_from = Vec::new();
-            let mut compact_to_set = HashMap::new();
-            for layer in &compact_to {
-                compact_to_set.insert(layer.layer_desc().key(), layer);
-            }
-            for layer in &layer_selection {
-                if let Some(to) = compact_to_set.get(&layer.layer_desc().key()) {
-                    tracing::info!(
-                        "skipping delete {} because found same layer key at different generation {}",
-                        layer, to
-                    );
-                } else {
-                    compact_from.push(layer.clone());
-                }
-            }
-            compact_from
-        };
         self.remote_client
-            .schedule_compaction_update(&compact_from, &compact_to)?;
+            .schedule_compaction_update(&layer_selection, &compact_to)?;
 
         drop(gc_lock);
 

pageserver/src/tenant/timeline/layer_manager.rs

@@ -337,45 +337,16 @@ impl OpenLayerManager {
         compact_to: &[ResidentLayer],
         metrics: &TimelineMetrics,
     ) {
-        // gc-compaction could contain layer rewrites. We need to delete the old layers and insert the new ones.
-
-        // Match the old layers with the new layers
-        let mut add_layers = HashMap::new();
-        let mut rewrite_layers = HashMap::new();
-        let mut drop_layers = HashMap::new();
-        for layer in compact_from {
-            drop_layers.insert(layer.layer_desc().key(), layer.clone());
-        }
-        for layer in compact_to {
-            if let Some(old_layer) = drop_layers.remove(&layer.layer_desc().key()) {
-                rewrite_layers.insert(layer.layer_desc().key(), (old_layer.clone(), layer.clone()));
-            } else {
-                add_layers.insert(layer.layer_desc().key(), layer.clone());
-            }
-        }
-        let add_layers = add_layers.values().cloned().collect::<Vec<_>>();
-        let drop_layers = drop_layers.values().cloned().collect::<Vec<_>>();
-        let rewrite_layers = rewrite_layers.values().cloned().collect::<Vec<_>>();
-
-        self.rewrite_layers_inner(&rewrite_layers, &drop_layers, &add_layers, metrics);
+        // We can simply reuse compact l0 logic. Use a different function name to indicate a different type of layer map modification.
+        self.finish_compact_l0(compact_from, compact_to, metrics)
     }
 
     /// Called post-compaction when some previous generation image layers were trimmed.
-    pub fn rewrite_layers(
+    pub(crate) fn rewrite_layers(
         &mut self,
         rewrite_layers: &[(Layer, ResidentLayer)],
         drop_layers: &[Layer],
         metrics: &TimelineMetrics,
-    ) {
-        self.rewrite_layers_inner(rewrite_layers, drop_layers, &[], metrics);
-    }
-
-    fn rewrite_layers_inner(
-        &mut self,
-        rewrite_layers: &[(Layer, ResidentLayer)],
-        drop_layers: &[Layer],
-        add_layers: &[ResidentLayer],
-        metrics: &TimelineMetrics,
     ) {
         let mut updates = self.layer_map.batch_update();
         for (old_layer, new_layer) in rewrite_layers {
@@ -411,10 +382,6 @@ impl OpenLayerManager {
         for l in drop_layers {
             Self::delete_historic_layer(l, &mut updates, &mut self.layer_fmgr);
         }
-        for l in add_layers {
-            Self::insert_historic_layer(l.as_ref().clone(), &mut updates, &mut self.layer_fmgr);
-            metrics.record_new_file_metrics(l.layer_desc().file_size);
-        }
         updates.flush();
     }
 

pgxn/neon/neon.c

@@ -19,7 +19,6 @@
 #include "access/xlogrecovery.h"
 #endif
 #include "replication/logical.h"
-#include "replication/logicallauncher.h"
 #include "replication/slot.h"
 #include "replication/walsender.h"
 #include "storage/proc.h"
@@ -435,15 +434,6 @@ _PG_init(void)
 
 	restore_running_xacts_callback = RestoreRunningXactsFromClog;
 
-	DefineCustomBoolVariable(
-							"neon.disable_logical_replication_subscribers",
-							"Disables incomming logical replication",
-							NULL,
-							&disable_logical_replication_subscribers,
-							false,
-							PGC_SIGHUP,
-							0,
-							NULL, NULL, NULL);
 
 	DefineCustomBoolVariable(
 							"neon.allow_replica_misconfig",

safekeeper/client/src/mgmt_api.rs

@@ -4,7 +4,7 @@
 //! united.
 
 use reqwest::{IntoUrl, Method, StatusCode};
-use safekeeper_api::models::{TimelineCreateRequest, TimelineStatus};
+use safekeeper_api::models::TimelineStatus;
 use std::error::Error as _;
 use utils::{
     http::error::HttpErrorBody,
@@ -76,28 +76,6 @@ impl Client {
         }
     }
 
-    pub async fn create_timeline(&self, req: &TimelineCreateRequest) -> Result<TimelineStatus> {
-        let uri = format!(
-            "{}/v1/tenant/{}/timeline/{}",
-            self.mgmt_api_endpoint, req.tenant_id, req.timeline_id
-        );
-        let resp = self.post(&uri, req).await?;
-        resp.json().await.map_err(Error::ReceiveBody)
-    }
-
-    pub async fn delete_timeline(
-        &self,
-        tenant_id: TenantId,
-        timeline_id: TimelineId,
-    ) -> Result<TimelineStatus> {
-        let uri = format!(
-            "{}/v1/tenant/{}/timeline/{}",
-            self.mgmt_api_endpoint, tenant_id, timeline_id
-        );
-        let resp = self.request(Method::DELETE, &uri, ()).await?;
-        resp.json().await.map_err(Error::ReceiveBody)
-    }
-
     pub async fn timeline_status(
         &self,
         tenant_id: TenantId,
@@ -129,14 +107,6 @@ impl Client {
         self.get(&uri).await
     }
 
-    async fn post<B: serde::Serialize, U: IntoUrl>(
-        &self,
-        uri: U,
-        body: B,
-    ) -> Result<reqwest::Response> {
-        self.request(Method::POST, uri, body).await
-    }
-
     async fn get<U: IntoUrl>(&self, uri: U) -> Result<reqwest::Response> {
         self.request(Method::GET, uri, ()).await
     }

scripts/comment-test-report.js

@@ -84,12 +84,6 @@ const parseReportJson = async ({ reportJsonUrl, fetch }) => {
                 } else {
                     arch = "unknown"
                 }
-                let lfcState = ""
-                if (test.parameters.includes("'with-lfc'")) {
-                    lfcState = "with-lfc"
-                } else {
-                    lfcState = "without-lfc"
-                }
 
                 // Removing build type and PostgreSQL version from the test name to make it shorter
                 const testName = test.name.replace(new RegExp(`${buildType}-pg${pgVersion}-?`), "").replace("[]", "")
@@ -97,7 +91,6 @@ const parseReportJson = async ({ reportJsonUrl, fetch }) => {
                 test.pgVersion = pgVersion
                 test.buildType = buildType
                 test.arch = arch
-                test.lfcState = lfcState
 
                 if (test.status === "passed") {
                     passedTests[pgVersion][testName].push(test)
@@ -164,7 +157,7 @@ const reportSummary = async (params) => {
                 const links = []
                 for (const test of tests) {
                     const allureLink = `${reportUrl}#suites/${test.parentUid}/${test.uid}`
-                    links.push(`[${test.buildType}-${test.arch}-${test.lfcState}](${allureLink})`)
+                    links.push(`[${test.buildType}-${test.arch}](${allureLink})`)
                 }
                 summary += `- \`${testName}\`: ${links.join(", ")}\n`
             }
@@ -195,7 +188,7 @@ const reportSummary = async (params) => {
                     const links = []
                     for (const test of tests) {
                         const allureLink = `${reportUrl}#suites/${test.parentUid}/${test.uid}/retries`
-                        links.push(`[${test.buildType}-${test.arch}-${test.lfcState}](${allureLink})`)
+                        links.push(`[${test.buildType}-${test.arch}](${allureLink})`)
                     }
                     summary += `- \`${testName}\`: ${links.join(", ")}\n`
                 }

scripts/ingest_regress_test_result-new-format.py

@@ -134,7 +134,7 @@ def ingest_test_result(
             if p["name"].startswith("__")
         }
         arch = parameters.get("arch", "UNKNOWN").strip("'")
-        lfc = parameters.get("lfc", "without-lfc").strip("'") == "with-lfc"
+        lfc = parameters.get("lfc", "False") == "True"
 
         build_type, pg_version, unparametrized_name = parse_test_name(test["name"])
         labels = {label["name"]: label["value"] for label in test["labels"]}

test_runner/fixtures/neon_cli.py

@@ -523,7 +523,6 @@ class NeonLocalCli(AbstractNeonCli):
         remote_ext_config: str | None = None,
         pageserver_id: int | None = None,
         allow_multiple: bool = False,
-        create_test_user: bool = False,
         basebackup_request_tries: int | None = None,
         env: dict[str, str] | None = None,
     ) -> subprocess.CompletedProcess[str]:
@@ -545,8 +544,6 @@ class NeonLocalCli(AbstractNeonCli):
             args.extend(["--pageserver-id", str(pageserver_id)])
         if allow_multiple:
             args.extend(["--allow-multiple"])
-        if create_test_user:
-            args.extend(["--create-test-user"])
 
         res = self.raw_cli(args, extra_env_vars)
         res.check_returncode()

test_runner/fixtures/neon_fixtures.py

@@ -3918,7 +3918,6 @@ class Endpoint(PgProtocol, LogUtils):
         pageserver_id: int | None = None,
         safekeepers: list[int] | None = None,
         allow_multiple: bool = False,
-        create_test_user: bool = False,
         basebackup_request_tries: int | None = None,
         env: dict[str, str] | None = None,
     ) -> Self:
@@ -3940,7 +3939,6 @@ class Endpoint(PgProtocol, LogUtils):
             remote_ext_config=remote_ext_config,
             pageserver_id=pageserver_id,
             allow_multiple=allow_multiple,
-            create_test_user=create_test_user,
             basebackup_request_tries=basebackup_request_tries,
             env=env,
         )

test_runner/fixtures/parametrize.py

@@ -121,8 +121,6 @@ def pytest_runtest_makereport(*args, **kwargs):
     }.get(os.uname().machine, "UNKNOWN")
     arch = os.getenv("RUNNER_ARCH", uname_m)
     allure.dynamic.parameter("__arch", arch)
-    allure.dynamic.parameter(
-        "__lfc", "with-lfc" if os.getenv("USE_LFC") != "false" else "without-lfc"
-    )
+    allure.dynamic.parameter("__lfc", os.getenv("USE_LFC") != "false")
 
     yield

test_runner/regress/test_compaction.py

@@ -1,8 +1,6 @@
 from __future__ import annotations
 
 import json
-import math
-import random
 import time
 from enum import StrEnum
 
@@ -130,6 +128,11 @@ def test_pageserver_gc_compaction_smoke(neon_env_builder: NeonEnvBuilder, with_b
     }
 
     env = neon_env_builder.init_start(initial_tenant_conf=SMOKE_CONF)
+    env.pageserver.allowed_errors.append(
+        r".*failed to acquire partition lock during gc-compaction.*"
+    )
+    env.pageserver.allowed_errors.append(r".*repartition() called concurrently.*")
+
     tenant_id = env.initial_tenant
     timeline_id = env.initial_timeline
 
@@ -144,14 +147,11 @@ def test_pageserver_gc_compaction_smoke(neon_env_builder: NeonEnvBuilder, with_b
     log.info("Writing initial data ...")
     workload.write_rows(row_count, env.pageserver.id)
 
-    ps_http.timeline_gc(
-        tenant_id, timeline_id, None
-    )  # Force refresh gc info to have gc_cutoff generated
-
     child_workloads: list[Workload] = []
 
     for i in range(1, churn_rounds + 1):
-        log.info(f"Running churn round {i}/{churn_rounds} ...")
+        if i % 10 == 0:
+            log.info(f"Running churn round {i}/{churn_rounds} ...")
         if i % 10 == 5 and with_branches == "with_branches":
             branch_name = f"child-{i}"
             branch_timeline_id = env.create_branch(branch_name)
@@ -172,241 +172,12 @@ def test_pageserver_gc_compaction_smoke(neon_env_builder: NeonEnvBuilder, with_b
                     "sub_compaction_max_job_size_mb": 16,
                 },
             )
-        # do not wait for upload so that we can see if gc_compaction works well with data being ingested
-        workload.churn_rows(row_count, env.pageserver.id, upload=False)
-        time.sleep(1)
-        workload.validate(env.pageserver.id)
 
-    def compaction_finished():
-        queue_depth = len(ps_http.timeline_compact_info(tenant_id, timeline_id))
-        assert queue_depth == 0
-
-    wait_until(compaction_finished, timeout=60)
-
-    # ensure gc_compaction is scheduled and it's actually running (instead of skipping due to no layers picked)
-    env.pageserver.assert_log_contains(
-        "scheduled_compact_timeline.*picked .* layers for compaction"
-    )
-
-    log.info("Validating at workload end ...")
-    workload.validate(env.pageserver.id)
-    for child_workload in child_workloads:
-        log.info(f"Validating at branch {child_workload.branch_name}")
-        child_workload.validate(env.pageserver.id)
-
-    # Run a legacy compaction+gc to ensure gc-compaction can coexist with legacy compaction.
-    ps_http.timeline_checkpoint(tenant_id, timeline_id, wait_until_uploaded=True)
-    ps_http.timeline_gc(tenant_id, timeline_id, None)
-
-
-@pytest.mark.parametrize(
-    "compaction_mode",
-    ["before_restart", "after_restart"],
-)
-def test_pageserver_gc_compaction_idempotent(
-    neon_env_builder: NeonEnvBuilder, compaction_mode: str
-):
-    """
-    Do gc-compaction twice without writing any new data and see if anything breaks.
-    We run this test in two modes:
-    - before_restart: run two gc-compactions before pageserver restart
-    - after_restart: run one gc-compaction before and one after pageserver restart
-    """
-    SMOKE_CONF = {
-        # Run both gc and gc-compaction.
-        "gc_period": "5s",
-        "compaction_period": "5s",
-        # No PiTR interval and small GC horizon
-        "pitr_interval": "0s",
-        "gc_horizon": 1024,
-        "lsn_lease_length": "0s",
-    }
-
-    env = neon_env_builder.init_start(initial_tenant_conf=SMOKE_CONF)
-    tenant_id = env.initial_tenant
-    timeline_id = env.initial_timeline
-
-    # Only in testing mode: the warning is expected because we rewrite a layer file of different generations.
-    # We could potentially patch the sanity-check code to not emit the warning in the future.
-    env.pageserver.allowed_errors.append(".*was unlinked but was not dangling.*")
-
-    row_count = 10000
-
-    ps_http = env.pageserver.http_client()
-
-    workload = Workload(env, tenant_id, timeline_id)
-    workload.init(env.pageserver.id)
-
-    workload.write_rows(row_count, env.pageserver.id)
-
-    child_workloads: list[Workload] = []
-
-    def compaction_finished():
-        queue_depth = len(ps_http.timeline_compact_info(tenant_id, timeline_id))
-        assert queue_depth == 0
-
-    workload.churn_rows(row_count, env.pageserver.id)
-    env.create_branch("child_branch")  # so that we have a retain_lsn
-    workload.churn_rows(row_count, env.pageserver.id)
-    # compact 3 times if mode is before_restart
-    n_compactions = 3 if compaction_mode == "before_restart" else 1
-    for _ in range(n_compactions):
-        # Force refresh gc info to have gc_cutoff generated
-        ps_http.timeline_gc(tenant_id, timeline_id, None)
-        ps_http.timeline_compact(
-            tenant_id,
-            timeline_id,
-            enhanced_gc_bottom_most_compaction=True,
-            body={
-                "scheduled": True,
-                "sub_compaction": True,
-                "compact_key_range": {
-                    "start": "000000000000000000000000000000000000",
-                    "end": "030000000000000000000000000000000000",
-                },
-                "sub_compaction_max_job_size_mb": 16,
-            },
-        )
-        wait_until(compaction_finished, timeout=60)
-    if compaction_mode == "after_restart":
-        env.pageserver.restart(True)
-        ps_http.timeline_gc(
-            tenant_id, timeline_id, None
-        )  # Force refresh gc info to have gc_cutoff generated
-        for _ in range(3):
-            ps_http.timeline_compact(
-                tenant_id,
-                timeline_id,
-                enhanced_gc_bottom_most_compaction=True,
-                body={
-                    "scheduled": True,
-                    "sub_compaction": True,
-                    "compact_key_range": {
-                        "start": "000000000000000000000000000000000000",
-                        "end": "030000000000000000000000000000000000",
-                    },
-                    "sub_compaction_max_job_size_mb": 16,
-                },
-            )
-            wait_until(compaction_finished, timeout=60)
-
-    # ensure gc_compaction is scheduled and it's actually running (instead of skipping due to no layers picked)
-    env.pageserver.assert_log_contains(
-        "scheduled_compact_timeline.*picked .* layers for compaction"
-    )
-
-    # ensure we hit the duplicated layer key warning at least once: we did two compactions consecutively,
-    # and the second one should have hit the duplicated layer key warning.
-    if compaction_mode == "before_restart":
-        env.pageserver.assert_log_contains("duplicated layer key in the same generation")
-    else:
-        env.pageserver.assert_log_contains("same layer key at different generation")
-
-    log.info("Validating at workload end ...")
-    workload.validate(env.pageserver.id)
-    for child_workload in child_workloads:
-        log.info(f"Validating at branch {child_workload.branch_name}")
-        child_workload.validate(env.pageserver.id)
-
-    # Run a legacy compaction+gc to ensure gc-compaction can coexist with legacy compaction.
-    ps_http.timeline_checkpoint(tenant_id, timeline_id, wait_until_uploaded=True)
-    ps_http.timeline_gc(tenant_id, timeline_id, None)
-
-
-@skip_in_debug_build("only run with release build")
-def test_pageserver_gc_compaction_interrupt(neon_env_builder: NeonEnvBuilder):
-    """
-    Force interrupt a gc-compaction and see if anything breaks.
-    """
-    SMOKE_CONF = {
-        # Run both gc and gc-compaction.
-        "gc_period": "5s",
-        "compaction_period": "5s",
-        # No PiTR interval and small GC horizon
-        "pitr_interval": "0s",
-        "gc_horizon": "1024",
-        "lsn_lease_length": "0s",
-    }
-
-    env = neon_env_builder.init_start(initial_tenant_conf=SMOKE_CONF)
-    tenant_id = env.initial_tenant
-    timeline_id = env.initial_timeline
-
-    # Only in testing mode: the warning is expected because we rewrite a layer file of different generations.
-    # We could potentially patch the sanity-check code to not emit the warning in the future.
-    env.pageserver.allowed_errors.append(".*was unlinked but was not dangling.*")
-
-    row_count = 10000
-    churn_rounds = 20
-
-    ps_http = env.pageserver.http_client()
-
-    workload = Workload(env, tenant_id, timeline_id)
-    workload.init(env.pageserver.id)
-
-    log.info("Writing initial data ...")
-    workload.write_rows(row_count, env.pageserver.id)
-
-    def compaction_finished():
-        queue_depth = len(ps_http.timeline_compact_info(tenant_id, timeline_id))
-        assert queue_depth == 0
-
-    expected_compaction_time_seconds = 5.0
-    ps_http.timeline_gc(
-        tenant_id, timeline_id, None
-    )  # Force refresh gc info to have gc_cutoff generated
-    for i in range(1, churn_rounds + 1):
-        log.info(f"Running churn round {i}/{churn_rounds} ...")
         workload.churn_rows(row_count, env.pageserver.id)
-        ps_http.timeline_compact(
-            tenant_id,
-            timeline_id,
-            enhanced_gc_bottom_most_compaction=True,
-            body={
-                "scheduled": True,
-                "sub_compaction": True,
-                "compact_key_range": {
-                    "start": "000000000000000000000000000000000000",
-                    "end": "030000000000000000000000000000000000",
-                },
-                "sub_compaction_max_job_size_mb": 16,
-            },
-        )
-        # sleep random seconds between 0 and max(compaction_time); if the result is 0, wait until the compaction is complete
-        # This would hopefully trigger the restart at different periods of the compaction:
-        # - while we are doing the compaction
-        # - while we finished the compaction but not yet uploaded the metadata
-        # - after we uploaded the metadata
-        time_to_sleep = random.randint(0, max(5, math.ceil(expected_compaction_time_seconds)))
-        if time_to_sleep == 0 or i == 1:
-            start = time.time()
-            wait_until(compaction_finished, timeout=60)
-            end = time.time()
-            expected_compaction_time_seconds = end - start
-            log.info(
-                f"expected_compaction_time_seconds updated to {expected_compaction_time_seconds} seconds"
-            )
-        else:
-            time.sleep(time_to_sleep)
-        env.pageserver.restart(True)
-        ps_http.timeline_gc(
-            tenant_id, timeline_id, None
-        )  # Force refresh gc info to have gc_cutoff generated
-        ps_http.timeline_compact(
-            tenant_id,
-            timeline_id,
-            enhanced_gc_bottom_most_compaction=True,
-            body={
-                "scheduled": True,
-                "sub_compaction": True,
-                "compact_key_range": {
-                    "start": "000000000000000000000000000000000000",
-                    "end": "030000000000000000000000000000000000",
-                },
-                "sub_compaction_max_job_size_mb": 16,
-            },
-        )
-        workload.validate(env.pageserver.id)
+
+    def compaction_finished():
+        queue_depth = len(ps_http.timeline_compact_info(tenant_id, timeline_id))
+        assert queue_depth == 0
 
     wait_until(compaction_finished, timeout=60)
 
@@ -417,6 +188,9 @@ def test_pageserver_gc_compaction_interrupt(neon_env_builder: NeonEnvBuilder):
 
     log.info("Validating at workload end ...")
     workload.validate(env.pageserver.id)
+    for child_workload in child_workloads:
+        log.info(f"Validating at branch {child_workload.branch_name}")
+        child_workload.validate(env.pageserver.id)
 
     # Run a legacy compaction+gc to ensure gc-compaction can coexist with legacy compaction.
     ps_http.timeline_checkpoint(tenant_id, timeline_id, wait_until_uploaded=True)
@@ -463,9 +237,7 @@ def test_sharding_compaction(
         "pitr_interval": "0s",
         # disable background compaction and GC. We invoke it manually when we want it to happen.
         "gc_period": "0s",
-        "gc_horizon": f"{128 * 1024}",
         "compaction_period": "0s",
-        "lsn_lease_length": "0s",
         # create image layers eagerly: we want to exercise image layer creation in this test.
         "image_creation_threshold": "1",
         "image_layer_creation_check_threshold": 0,
@@ -540,8 +312,6 @@ def test_sharding_compaction(
         for shard in env.storage_controller.locate(tenant_id):
             pageserver = env.get_pageserver(shard["node_id"])
             tenant_shard_id = shard["shard_id"]
-            # Force refresh gc info to have gc_cutoff generated
-            pageserver.http_client().timeline_gc(tenant_shard_id, timeline_id, None)
             pageserver.http_client().timeline_compact(
                 tenant_shard_id,
                 timeline_id,

test_runner/regress/test_storage_scrubber.py

@@ -271,14 +271,8 @@ def test_scrubber_physical_gc_ancestors(neon_env_builder: NeonEnvBuilder, shard_
         ps.http_client().timeline_compact(
             shard, timeline_id, force_image_layer_creation=True, wait_until_uploaded=True
         )
-
-    # Add some WAL so that we don't gc at the latest remote consistent lsn
-    workload.churn_rows(10)
-
-    # Now gc the old stuff away
-    for shard in shards:
-        ps = env.get_tenant_pageserver(shard)
-        assert ps is not None
+        # Add some WAL so that we don't gc at the latest remote consistent lsn
+        workload.churn_rows(1)
         ps.http_client().timeline_gc(shard, timeline_id, 0)
 
     # We will use a min_age_secs=1 threshold for deletion, let it pass

test_runner/regress/test_subscriber_branching.py

@@ -1,242 +0,0 @@
-from __future__ import annotations
-
-import time
-
-from fixtures.log_helper import log
-from fixtures.neon_fixtures import NeonEnv, logical_replication_sync
-from fixtures.utils import query_scalar, wait_until
-
-
-# This test checks that branching of timeline with logical subscriptions
-# does not affect logical replication for parent.
-# Endpoint on a new branch will drop all existing subscriptions at the start,
-# so it will not receive any changes.
-# If needed, user can create new subscriptions on the child branch.
-def test_subscriber_branching(neon_simple_env: NeonEnv):
-    env = neon_simple_env
-    env.create_branch("publisher")
-    pub = env.endpoints.create("publisher")
-    pub.respec(
-        skip_pg_catalog_updates=False,
-        create_test_user=True,
-    )
-    pub.start(create_test_user=True)
-
-    env.create_branch("subscriber")
-    sub = env.endpoints.create("subscriber")
-    # Pass create_test_user flag to get properly filled spec.users and spec.databases fields.
-    #
-    # This test checks the per-database operations that happen at compute start
-    # and these operations are applied to the databases that are present in the spec.
-    sub.respec(
-        skip_pg_catalog_updates=False,
-        create_test_user=True,
-    )
-    sub.start(create_test_user=True)
-
-    pub.wait_for_migrations()
-    sub.wait_for_migrations()
-
-    n_records = 1000
-
-    def check_that_changes_propagated():
-        scur.execute("SELECT count(*) FROM t")
-        res = scur.fetchall()
-        assert res[0][0] == n_records
-
-    def insert_data(pub, start):
-        with pub.cursor(dbname="neondb", user="test", password="pubtestpwd") as pcur:
-            for i in range(start, start + n_records):
-                pcur.execute("INSERT into t values (%s,random()*100000)", (i,))
-
-    # create_test_user creates a user without password
-    # but psycopg2 execute() requires a password
-    with sub.cursor() as scur:
-        scur.execute("ALTER USER test WITH PASSWORD 'testpwd'")
-    with pub.cursor() as pcur:
-        # Create a test user to avoid using superuser
-        pcur.execute("ALTER USER test WITH PASSWORD 'pubtestpwd'")
-        # If we don't do this, creating the subscription will fail
-        pub.edit_hba(["host all test 0.0.0.0/0 md5"])
-
-    with pub.cursor(dbname="neondb", user="test", password="pubtestpwd") as pcur:
-        pcur.execute("CREATE TABLE t (pk integer primary key, sk integer)")
-        pcur.execute("CREATE PUBLICATION pub FOR TABLE t")
-
-        with sub.cursor(dbname="neondb", user="test", password="testpwd") as scur:
-            scur.execute("CREATE TABLE t (pk integer primary key, sk integer)")
-            pub_conn = (
-                f"host=localhost port={pub.pg_port} dbname=neondb user=test password=pubtestpwd"
-            )
-            query = f"CREATE SUBSCRIPTION sub CONNECTION '{pub_conn}' PUBLICATION pub"
-            scur.execute(query)
-            time.sleep(2)  # let initial table sync complete
-
-        insert_data(pub, 0)
-
-        with sub.cursor(dbname="neondb", user="test", password="testpwd") as scur:
-            wait_until(check_that_changes_propagated)
-            latest_end_lsn = query_scalar(
-                scur, "select latest_end_lsn from pg_catalog.pg_stat_subscription; "
-            )
-            last_insert_lsn = query_scalar(scur, "select pg_current_wal_insert_lsn();")
-
-            log.info(f"latest_end_lsn = {latest_end_lsn}")
-            log.info(f"last_insert_lsn = {last_insert_lsn}")
-
-        # stop the parent subscriber so that it doesn't interfere with the test
-        sub.stop()
-
-        # 1. good scenario:
-        # create subscriber_child_1
-        # it will not get changes from publisher, because drop_subscriptions_before_start is set to True
-        sub_child_1_timeline_id = env.create_branch(
-            "subscriber_child_1",
-            ancestor_branch_name="subscriber",
-            ancestor_start_lsn=last_insert_lsn,
-        )
-        sub_child_1 = env.endpoints.create("subscriber_child_1")
-        # Pass drop_subscriptions_before_start flag
-        sub_child_1.respec(
-            skip_pg_catalog_updates=False,
-            create_test_user=True,
-            drop_subscriptions_before_start=True,
-        )
-        sub_child_1.start(create_test_user=True)
-
-        # ensure that subscriber_child_1 sees all the data
-        with sub_child_1.cursor(dbname="neondb", user="test", password="testpwd") as scur:
-            scur.execute("SELECT count(*) FROM t")
-            res = scur.fetchall()
-            assert res[0][0] == n_records
-
-            # ensure that there are no subscriptions in this database
-            scur.execute("SELECT 1 FROM pg_catalog.pg_subscription WHERE subname = 'sub'")
-            assert len(scur.fetchall()) == 0
-
-        # ensure that drop_subscriptions_done happened on this timeline
-        with sub_child_1.cursor() as scur_postgres:
-            scur_postgres.execute("SELECT timeline_id from neon.drop_subscriptions_done")
-            res = scur_postgres.fetchall()
-            assert len(res) == 1
-            assert str(sub_child_1_timeline_id) == res[0][0]
-
-        old_n_records = n_records
-        # insert more data on publisher
-        insert_data(pub, n_records)
-        n_records += n_records
-
-        pcur.execute("SELECT count(*) FROM t")
-        res = pcur.fetchall()
-        assert res[0][0] == n_records
-
-        # ensure that subscriber_child_1 doesn't see the new data
-        with sub_child_1.cursor(dbname="neondb", user="test", password="testpwd") as scur:
-            scur.execute("SELECT count(*) FROM t")
-            res = scur.fetchall()
-            assert res[0][0] == old_n_records
-
-        # reenable logical replication on subscriber_child_1
-        # using new publication
-        # ensure that new publication works as expected
-        with sub_child_1.cursor(dbname="neondb", user="test", password="testpwd") as scur:
-            scur.execute("TRUNCATE t")
-
-            # create new subscription
-            # with new pub name
-            pcur.execute("CREATE PUBLICATION pub_new FOR TABLE t")
-            query = f"CREATE SUBSCRIPTION sub_new CONNECTION '{pub_conn}' PUBLICATION pub_new"
-            scur.execute(query)
-
-            wait_until(check_that_changes_propagated)
-
-            scur.execute("SELECT count(*) FROM t")
-            res = scur.fetchall()
-            assert res[0][0] == n_records
-
-        # ensure that new publication works as expected after compute restart
-        # first restart with drop_subscriptions_before_start=True
-        # to emulate the case when compute restarts within the VM with stale spec
-        sub_child_1.stop()
-        sub_child_1.respec(
-            skip_pg_catalog_updates=False,
-            create_test_user=True,
-            drop_subscriptions_before_start=True,
-        )
-        sub_child_1.start(create_test_user=True)
-
-        with sub_child_1.cursor(dbname="neondb", user="test", password="testpwd") as scur:
-            # ensure that even though the flag is set, we didn't drop new subscription
-            scur.execute("SELECT 1 FROM pg_catalog.pg_subscription WHERE subname = 'sub_new'")
-            assert len(scur.fetchall()) == 1
-
-        # ensure that drop_subscriptions_done happened on this timeline
-        with sub_child_1.cursor() as scur_postgres:
-            scur_postgres.execute("SELECT timeline_id from neon.drop_subscriptions_done")
-            res = scur_postgres.fetchall()
-            assert len(res) == 1
-            assert str(sub_child_1_timeline_id) == res[0][0]
-
-        sub_child_1.stop()
-        sub_child_1.respec(
-            skip_pg_catalog_updates=False,
-            create_test_user=True,
-            drop_subscriptions_before_start=False,
-        )
-        sub_child_1.start(create_test_user=True)
-
-        # insert more data on publisher
-        insert_data(pub, n_records)
-        n_records += n_records
-        with sub_child_1.cursor(dbname="neondb", user="test", password="testpwd") as scur:
-            # ensure that there is a subscriptions in this database
-            scur.execute("SELECT 1 FROM pg_catalog.pg_subscription WHERE subname = 'sub_new'")
-            assert len(scur.fetchall()) == 1
-
-            wait_until(check_that_changes_propagated)
-            scur.execute("SELECT count(*) FROM t")
-            res = scur.fetchall()
-            assert res[0][0] == n_records
-
-        # ensure that drop_subscriptions_done happened on this timeline
-        with sub_child_1.cursor() as scur_postgres:
-            scur_postgres.execute("SELECT timeline_id from neon.drop_subscriptions_done")
-            res = scur_postgres.fetchall()
-            assert len(res) == 1
-            assert str(sub_child_1_timeline_id) == res[0][0]
-
-        # wake the sub and ensure that it catches up with the new data
-        sub.start(create_test_user=True)
-        with sub.cursor(dbname="neondb", user="test", password="testpwd") as scur:
-            logical_replication_sync(sub, pub)
-            wait_until(check_that_changes_propagated)
-            scur.execute("SELECT count(*) FROM t")
-            res = scur.fetchall()
-            assert res[0][0] == n_records
-
-        # test that we can create a branch of a branch
-        sub_child_2_timeline_id = env.create_branch(
-            "subscriber_child_2",
-            ancestor_branch_name="subscriber_child_1",
-        )
-        sub_child_2 = env.endpoints.create("subscriber_child_2")
-        # Pass drop_subscriptions_before_start flag
-        sub_child_2.respec(
-            skip_pg_catalog_updates=False,
-            drop_subscriptions_before_start=True,
-        )
-        sub_child_2.start(create_test_user=True)
-
-        # ensure that subscriber_child_2 does not inherit subscription from child_1
-        with sub_child_2.cursor(dbname="neondb", user="test", password="testpwd") as scur:
-            # ensure that there are no subscriptions in this database
-            scur.execute("SELECT count(*) FROM pg_catalog.pg_subscription")
-            res = scur.fetchall()
-            assert res[0][0] == 0
-
-        # ensure that drop_subscriptions_done happened on this timeline
-        with sub_child_2.cursor() as scur_postgres:
-            scur_postgres.execute("SELECT timeline_id from neon.drop_subscriptions_done")
-            res = scur_postgres.fetchall()
-            assert len(res) == 1
-            assert str(sub_child_2_timeline_id) == res[0][0]

test_runner/regress/test_timeline_detach_ancestor.py

@@ -607,7 +607,7 @@ def test_timeline_ancestor_detach_idempotent_success(
 
     if shards_after > 1:
         # FIXME: should this be in the neon_env_builder.init_start?
-        env.storage_controller.reconcile_until_idle(timeout_secs=120)
+        env.storage_controller.reconcile_until_idle()
         client = env.storage_controller.pageserver_api()
     else:
         client = env.pageserver.http_client()
@@ -636,7 +636,7 @@ def test_timeline_ancestor_detach_idempotent_success(
         # Do a shard split
         # This is a reproducer for https://github.com/neondatabase/neon/issues/9667
         env.storage_controller.tenant_shard_split(env.initial_tenant, shards_after)
-        env.storage_controller.reconcile_until_idle(timeout_secs=120)
+        env.storage_controller.reconcile_until_idle()
 
     first_reparenting_response = client.detach_ancestor(env.initial_tenant, first_branch)
     assert set(first_reparenting_response) == {reparented1, reparented2}

vendor/revisions.json

@@ -1,18 +1,18 @@
 {
   "v17": [
     "17.2",
-    "4276717f6e91023e504de355f4f21d4824074de8"
+    "a8dd6e779dde907778006adb436b557ad652fb97"
   ],
   "v16": [
     "16.6",
-    "061d56377961ba56998e41b7d5d5e975919ad301"
+    "d674efd776f59d78e8fa1535bd2f95c3e6984fca"
   ],
   "v15": [
     "15.10",
-    "935292e883298187f112db6e9c7f765037ddcf64"
+    "dd0b28d6fbad39e227f3b77296fcca879af8b3a9"
   ],
   "v14": [
     "14.15",
-    "5f3b3afdd7c24b4a0fd63ecb3288fab472fcc633"
+    "46082f20884f087a2d974b33ac65d63af26142bd"
   ]
 }