Fix construction of database URL

.github/workflows/build_and_test.yml

@@ -852,7 +852,7 @@ jobs:
       run:
         shell: sh -eu {0}
     env:
-      VM_BUILDER_VERSION: v0.19.0
+      VM_BUILDER_VERSION: v0.18.5
 
     steps:
       - name: Checkout
@@ -874,7 +874,8 @@ jobs:
       - name: Build vm image
         run: |
           ./vm-builder \
-            -spec=vm-image-spec.yaml \
+            -enable-file-cache \
+            -cgroup-uid=postgres \
             -src=369495373322.dkr.ecr.eu-central-1.amazonaws.com/compute-node-${{ matrix.version }}:${{needs.tag.outputs.build-tag}} \
             -dst=369495373322.dkr.ecr.eu-central-1.amazonaws.com/vm-compute-node-${{ matrix.version }}:${{needs.tag.outputs.build-tag}}
 

Cargo.lock

@@ -3221,7 +3221,7 @@ dependencies = [
 [[package]]
 name = "postgres"
 version = "0.19.4"
-source = "git+https://github.com/neondatabase/rust-postgres.git?rev=6ce32f791526e27533cab0232a6bb243b2c32584#6ce32f791526e27533cab0232a6bb243b2c32584"
+source = "git+https://github.com/neondatabase/rust-postgres.git?rev=ce7260db5998fe27167da42503905a12e7ad9048#ce7260db5998fe27167da42503905a12e7ad9048"
 dependencies = [
  "bytes",
  "fallible-iterator",
@@ -3234,7 +3234,7 @@ dependencies = [
 [[package]]
 name = "postgres-native-tls"
 version = "0.5.0"
-source = "git+https://github.com/neondatabase/rust-postgres.git?rev=6ce32f791526e27533cab0232a6bb243b2c32584#6ce32f791526e27533cab0232a6bb243b2c32584"
+source = "git+https://github.com/neondatabase/rust-postgres.git?rev=ce7260db5998fe27167da42503905a12e7ad9048#ce7260db5998fe27167da42503905a12e7ad9048"
 dependencies = [
  "native-tls",
  "tokio",
@@ -3245,7 +3245,7 @@ dependencies = [
 [[package]]
 name = "postgres-protocol"
 version = "0.6.4"
-source = "git+https://github.com/neondatabase/rust-postgres.git?rev=6ce32f791526e27533cab0232a6bb243b2c32584#6ce32f791526e27533cab0232a6bb243b2c32584"
+source = "git+https://github.com/neondatabase/rust-postgres.git?rev=ce7260db5998fe27167da42503905a12e7ad9048#ce7260db5998fe27167da42503905a12e7ad9048"
 dependencies = [
  "base64 0.20.0",
  "byteorder",
@@ -3263,7 +3263,7 @@ dependencies = [
 [[package]]
 name = "postgres-types"
 version = "0.2.4"
-source = "git+https://github.com/neondatabase/rust-postgres.git?rev=6ce32f791526e27533cab0232a6bb243b2c32584#6ce32f791526e27533cab0232a6bb243b2c32584"
+source = "git+https://github.com/neondatabase/rust-postgres.git?rev=ce7260db5998fe27167da42503905a12e7ad9048#ce7260db5998fe27167da42503905a12e7ad9048"
 dependencies = [
  "bytes",
  "fallible-iterator",
@@ -4933,7 +4933,7 @@ dependencies = [
 [[package]]
 name = "tokio-postgres"
 version = "0.7.7"
-source = "git+https://github.com/neondatabase/rust-postgres.git?rev=6ce32f791526e27533cab0232a6bb243b2c32584#6ce32f791526e27533cab0232a6bb243b2c32584"
+source = "git+https://github.com/neondatabase/rust-postgres.git?rev=ce7260db5998fe27167da42503905a12e7ad9048#ce7260db5998fe27167da42503905a12e7ad9048"
 dependencies = [
  "async-trait",
  "byteorder",

Cargo.toml

@@ -165,11 +165,11 @@ env_logger = "0.10"
 log = "0.4"
 
 ## Libraries from neondatabase/ git forks, ideally with changes to be upstreamed
-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev="6ce32f791526e27533cab0232a6bb243b2c32584" }
-postgres-native-tls = { git = "https://github.com/neondatabase/rust-postgres.git", rev="6ce32f791526e27533cab0232a6bb243b2c32584" }
-postgres-protocol = { git = "https://github.com/neondatabase/rust-postgres.git", rev="6ce32f791526e27533cab0232a6bb243b2c32584" }
-postgres-types = { git = "https://github.com/neondatabase/rust-postgres.git", rev="6ce32f791526e27533cab0232a6bb243b2c32584" }
-tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev="6ce32f791526e27533cab0232a6bb243b2c32584" }
+postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev="ce7260db5998fe27167da42503905a12e7ad9048" }
+postgres-native-tls = { git = "https://github.com/neondatabase/rust-postgres.git", rev="ce7260db5998fe27167da42503905a12e7ad9048" }
+postgres-protocol = { git = "https://github.com/neondatabase/rust-postgres.git", rev="ce7260db5998fe27167da42503905a12e7ad9048" }
+postgres-types = { git = "https://github.com/neondatabase/rust-postgres.git", rev="ce7260db5998fe27167da42503905a12e7ad9048" }
+tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev="ce7260db5998fe27167da42503905a12e7ad9048" }
 
 ## Other git libraries
 heapless = { default-features=false, features=[], git = "https://github.com/japaric/heapless.git", rev = "644653bf3b831c6bb4963be2de24804acf5e5001" } # upstream release pending
@@ -206,7 +206,7 @@ tonic-build = "0.9"
 
 # This is only needed for proxy's tests.
 # TODO: we should probably fork `tokio-postgres-rustls` instead.
-tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev="6ce32f791526e27533cab0232a6bb243b2c32584" }
+tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev="ce7260db5998fe27167da42503905a12e7ad9048" }
 
 ################# Binary contents sections
 

Dockerfile.compute-node

@@ -714,23 +714,6 @@ RUN wget https://github.com/pksunkara/pgx_ulid/archive/refs/tags/v0.1.3.tar.gz -
     cargo pgrx install --release && \
     echo "trusted = true" >> /usr/local/pgsql/share/extension/ulid.control
 
-#########################################################################################
-#
-# Layer "pg-wait-sampling-pg-build"
-# compile pg_wait_sampling extension
-#
-#########################################################################################
-FROM build-deps AS pg-wait-sampling-pg-build
-COPY --from=pg-build /usr/local/pgsql/ /usr/local/pgsql/
-
-ENV PATH "/usr/local/pgsql/bin/:$PATH"
-RUN wget https://github.com/postgrespro/pg_wait_sampling/archive/refs/tags/v1.1.5.tar.gz -O pg_wait_sampling.tar.gz && \
-    echo 'a03da6a413f5652ce470a3635ed6ebba528c74cb26aa4cfced8aff8a8441f81ec6dd657ff62cd6ce96a4e6ce02cad9f2519ae9525367ece60497aa20faafde5c  pg_wait_sampling.tar.gz' | sha512sum -c && \
-    mkdir pg_wait_sampling-src && cd pg_wait_sampling-src && tar xvzf ../pg_wait_sampling.tar.gz --strip-components=1 -C . && \
-    make USE_PGXS=1 -j $(getconf _NPROCESSORS_ONLN) && \
-    make USE_PGXS=1 -j $(getconf _NPROCESSORS_ONLN) install && \
-    echo 'trusted = true' >> /usr/local/pgsql/share/extension/pg_wait_sampling.control
-
 #########################################################################################
 #
 # Layer "neon-pg-ext-build"
@@ -767,7 +750,6 @@ COPY --from=rdkit-pg-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg-uuidv7-pg-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg-roaringbitmap-pg-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg-embedding-pg-build /usr/local/pgsql/ /usr/local/pgsql/
-COPY --from=pg-wait-sampling-pg-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY pgxn/ pgxn/
 
 RUN make -j $(getconf _NPROCESSORS_ONLN) \

compute_tools/src/compute.rs

@@ -2,7 +2,6 @@ use std::collections::HashMap;
 use std::env;
 use std::fs;
 use std::io::BufRead;
-use std::io::Write;
 use std::os::unix::fs::PermissionsExt;
 use std::path::Path;
 use std::process::{Command, Stdio};
@@ -15,7 +14,6 @@ use chrono::{DateTime, Utc};
 use futures::future::join_all;
 use futures::stream::FuturesUnordered;
 use futures::StreamExt;
-use notify::event;
 use postgres::{Client, NoTls};
 use tokio;
 use tokio_postgres;
@@ -85,6 +83,7 @@ pub struct ComputeState {
     pub last_active: Option<DateTime<Utc>>,
     pub error: Option<String>,
     pub pspec: Option<ParsedSpec>,
+    pub merge_src_connstr: Option<String>,
     pub metrics: ComputeMetrics,
 }
 
@@ -96,6 +95,7 @@ impl ComputeState {
             last_active: None,
             error: None,
             pspec: None,
+            merge_src_connstr: None,
             metrics: ComputeMetrics::default(),
         }
     }
@@ -629,6 +629,22 @@ impl ComputeNode {
         Ok(())
     }
 
+    /// Merge two branches
+    #[instrument(skip_all)]
+    pub fn merge(&self, src_connstr: &str) -> Result<()> {
+        let mut client = Client::connect(self.connstr.as_str(), NoTls)?;
+        handle_merge(&mut client, self.connstr.as_str(), &src_connstr)?;
+        Ok(())
+    }
+
+    /// Mark brnach as mergeable
+    #[instrument(skip_all)]
+    pub fn set_mergeable(&self) -> Result<()> {
+        let mut client = Client::connect(self.connstr.as_str(), NoTls)?;
+        handle_set_mergeable(&mut client, self.connstr.as_str())?;
+        Ok(())
+    }
+
     /// Start Postgres as a child process and manage DBs/roles.
     /// After that this will hang waiting on the postmaster process to exit.
     #[instrument(skip_all)]
@@ -646,30 +662,9 @@ impl ComputeNode {
             } else {
                 vec![]
             })
-            .stderr(Stdio::piped())
             .spawn()
             .expect("cannot start postgres process");
 
-        let stderr = pg.stderr.take().unwrap();
-        std::thread::spawn(move || {
-            let reader = std::io::BufReader::new(stderr);
-            let mut last_lines = vec![];
-            for line in reader.lines() {
-                if let Ok(line) = line {
-                    if line.starts_with("2023-") {
-                        // print all lines from the previous postgres instance
-                        let combined = format!("PG:{}\n", last_lines.join("\u{200B}"));
-                        let res = std::io::stderr().lock().write_all(combined.as_bytes());
-                        if let Err(e) = res {
-                            error!("failed to write to stderr: {}", e);
-                        }
-                        last_lines.clear();
-                    }
-                    last_lines.push(line);
-                }
-            }
-        });
-
         wait_for_postgres(&mut pg, pgdata_path)?;
 
         Ok(pg)

compute_tools/src/configurator.rs

@@ -31,6 +31,35 @@ fn configurator_main_loop(compute: &Arc<ComputeNode>) {
             // XXX: used to test that API is blocking
             // std::thread::sleep(std::time::Duration::from_millis(10000));
 
+            compute.set_status(new_status);
+        } else if state.status == ComputeStatus::MergePending {
+            info!("got merge request");
+            state.status = ComputeStatus::Merging;
+            compute.state_changed.notify_all();
+            let connstr = state.merge_src_connstr.clone().unwrap();
+            drop(state);
+
+            let mut new_status = ComputeStatus::Failed;
+            if let Err(e) = compute.merge(&connstr) {
+                info!("could not merge compute node: {}", e);
+            } else {
+                new_status = ComputeStatus::Running;
+                info!("merge complete");
+            }
+            compute.set_status(new_status);
+        } else if state.status == ComputeStatus::SetMergeablePending {
+            info!("got set mergeable request");
+            state.status = ComputeStatus::SetMergeable;
+            compute.state_changed.notify_all();
+            drop(state);
+
+            let mut new_status = ComputeStatus::Failed;
+            if let Err(e) = compute.set_mergeable() {
+                info!("could not mark branch as mergeable: {}", e);
+            } else {
+                new_status = ComputeStatus::Running;
+                info!("marked as mergeable");
+            }
             compute.set_status(new_status);
         } else if state.status == ComputeStatus::Failed {
             info!("compute node is now in Failed state, exiting");

compute_tools/src/http/api.rs

@@ -123,6 +123,30 @@ async fn routes(req: Request<Body>, compute: &Arc<ComputeNode>) -> Response<Body
             }
         }
 
+        // Handle branch merge request
+        (&Method::POST, "/merge") => {
+            info!("serving /merge POST request");
+            match handle_merge_request(req, compute).await {
+                Ok(msg) => Response::new(Body::from(msg)),
+                Err((msg, code)) => {
+                    error!("error handling /merge request: {msg}");
+                    render_json_error(&msg, code)
+                }
+            }
+        }
+
+        // Handle branch set mergeable request
+        (&Method::POST, "/set_mergeable") => {
+            info!("serving /set_mergeable POST request");
+            match handle_set_mergeable_request(compute).await {
+                Ok(msg) => Response::new(Body::from(msg)),
+                Err((msg, code)) => {
+                    error!("error handling /set_mergeable request: {msg}");
+                    render_json_error(&msg, code)
+                }
+            }
+        }
+
         // download extension files from S3 on demand
         (&Method::POST, route) if route.starts_with("/extension_server/") => {
             info!("serving {:?} POST request", route);
@@ -209,6 +233,103 @@ async fn routes(req: Request<Body>, compute: &Arc<ComputeNode>) -> Response<Body
     }
 }
 
+async fn handle_merge_request(
+    req: Request<Body>,
+    compute: &Arc<ComputeNode>,
+) -> Result<String, (String, StatusCode)> {
+    let body_bytes = hyper::body::to_bytes(req.into_body()).await.unwrap();
+    let connstr = String::from_utf8(body_bytes.to_vec()).unwrap();
+
+    let c = compute.clone();
+
+    {
+        let mut state = compute.state.lock().unwrap();
+        if state.status != ComputeStatus::Empty && state.status != ComputeStatus::Running {
+            let msg = format!(
+                "invalid compute status for merge request: {:?}",
+                state.status.clone()
+            );
+            return Err((msg, StatusCode::PRECONDITION_FAILED));
+        }
+        state.merge_src_connstr = Some(connstr);
+        state.status = ComputeStatus::MergePending;
+        compute.state_changed.notify_all();
+        drop(state);
+        info!("set new spec and notified waiters");
+    }
+
+    task::spawn_blocking(move || {
+        let mut state = c.state.lock().unwrap();
+        while state.status != ComputeStatus::Running {
+            state = c.state_changed.wait(state).unwrap();
+            info!(
+                "waiting for compute to become Running, current status: {:?}",
+                state.status
+            );
+
+            if state.status == ComputeStatus::Failed {
+                let err = state.error.as_ref().map_or("unknown error", |x| x);
+                let msg = format!("compute configuration failed: {:?}", err);
+                return Err((msg, StatusCode::INTERNAL_SERVER_ERROR));
+            }
+        }
+
+        Ok(())
+    })
+    .await
+    .unwrap()?;
+
+    let state = compute.state.lock().unwrap().clone();
+    let status_response = status_response_from_state(&state);
+    Ok(serde_json::to_string(&status_response).unwrap())
+}
+
+async fn handle_set_mergeable_request(
+    compute: &Arc<ComputeNode>,
+) -> Result<String, (String, StatusCode)> {
+    let c = compute.clone();
+
+    {
+        let mut state = compute.state.lock().unwrap();
+        if state.status != ComputeStatus::Empty && state.status != ComputeStatus::Running {
+            let msg = format!(
+                "invalid compute status for merge request: {:?}",
+                state.status.clone()
+            );
+            return Err((msg, StatusCode::PRECONDITION_FAILED));
+        }
+        state.status = ComputeStatus::SetMergeablePending;
+        compute.state_changed.notify_all();
+        drop(state);
+        info!("set new spec and notified waiters");
+    }
+
+    task::spawn_blocking(move || {
+        let mut state = c.state.lock().unwrap();
+        while state.status != ComputeStatus::Running {
+            state = c.state_changed.wait(state).unwrap();
+            info!(
+                "waiting for compute to become Running, current status: {:?}",
+                state.status
+            );
+
+            if state.status == ComputeStatus::Failed {
+                let err = state.error.as_ref().map_or("unknown error", |x| x);
+                let msg = format!("compute configuration failed: {:?}", err);
+                return Err((msg, StatusCode::INTERNAL_SERVER_ERROR));
+            }
+        }
+
+        Ok(())
+    })
+    .await
+    .unwrap()?;
+
+    let state = compute.state.lock().unwrap().clone();
+    let status_response = status_response_from_state(&state);
+    Ok(serde_json::to_string(&status_response).unwrap())
+}
+
 async fn handle_configure_request(
     req: Request<Body>,
     compute: &Arc<ComputeNode>,

compute_tools/src/http/openapi_spec.yaml

@@ -85,6 +85,61 @@ paths:
                 description: Error text or 'true' if check passed.
                 example: "true"
 
+  /merge:
+    post:
+      tags:
+      - Merge
+      summary: Merge branches.
+      description: |
+        This is a blocking API endpoint, i.e. it blocks waiting until
+        compute is finished configuration and is in `Running` state.
+        Optional non-blocking mode could be added later.
+      operationId: mergeBranches
+      requestBody:
+        description: connection string of target branch
+        required: true
+        content:
+          text/plain:
+            schema:
+              type: string
+      responses:
+        200:
+          description: Merge finished.
+          content:
+            application/json:OK
+              schema:
+                $ref: "#/components/schemas/ComputeState"
+        500:
+          description: Merge request failed.
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/GenericError"
+
+  /set_mergeable:
+    post:
+      tags:
+      - Set mergeable
+      summary: Mark branch as mergeable.
+      description: |
+        This is a blocking API endpoint, i.e. it blocks waiting until
+        compute is finished configuration and is in `Running` state.
+        Optional non-blocking mode could be added later.
+      operationId: markBranchMergeable
+      responses:
+        200:
+          description: Branch marked as mergeable
+          content:
+            application/json:OK
+              schema:
+                $ref: "#/components/schemas/ComputeState"
+        500:
+          description: Set mergeable request failed.
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/GenericError"
+
   /configure:
     post:
       tags:

compute_tools/src/spec.rs

@@ -407,6 +407,58 @@ fn reassign_owned_objects(spec: &ComputeSpec, connstr: &str, role_name: &PgIdent
     Ok(())
 }
 
+#[instrument(skip_all)]
+pub fn handle_merge(client: &mut Client, dst_connstr: &str, src_connstr: &str) -> Result<()> {
+    info!("Merge branch into {}", dst_connstr);
+    let existing_dbs = get_existing_dbs(client)?;
+
+    for (_, db) in existing_dbs {
+        if db.name.starts_with("template") {
+            continue;
+        }
+        let mut dst_conf = Config::from_str(dst_connstr)?;
+        dst_conf.dbname(&db.name);
+
+        let mut src_conf = Config::from_str(src_connstr)?;
+        src_conf.dbname(&db.name);
+
+        let mut sub_client = dst_conf.connect(NoTls)?;
+        let mut connstr_parts: Vec<&str> = src_connstr.split('/').collect();
+        connstr_parts.pop();
+        connstr_parts.push(&db.name);
+        let connstr = connstr_parts.join("/");
+        let create_sub = format!("create subscription sub_merge connection '{}' publication pub_merge with (create_slot=false, slot_name=merge_slot_{}, copy_data=false)", connstr, &db.name);
+        sub_client.simple_query(&create_sub)?;
+    }
+
+    Ok(())
+}
+
+#[instrument(skip_all)]
+pub fn handle_set_mergeable(client: &mut Client, connstr: &str) -> Result<()> {
+    info!("Mark branch as mergeable");
+    let existing_dbs = get_existing_dbs(client)?;
+
+    for (_, db) in existing_dbs {
+        if db.name.starts_with("template") {
+            continue;
+        }
+        let mut conf = Config::from_str(connstr)?;
+        conf.dbname(&db.name);
+
+        let mut db_client = conf.connect(NoTls)?;
+
+        let create_slot = format!(
+            "select pg_create_logical_replication_slot('merge_slot_{}', 'pgoutput')",
+            &db.name
+        );
+        db_client.simple_query(&create_slot)?;
+        db_client.simple_query("create publication pub_merge for all tables")?;
+    }
+
+    Ok(())
+}
+
 /// It follows mostly the same logic as `handle_roles()` excepting that we
 /// does not use an explicit transactions block, since major database operations
 /// like `CREATE DATABASE` and `DROP DATABASE` do not support it. Statement-level
@@ -670,12 +722,6 @@ pub fn handle_extensions(spec: &ComputeSpec, client: &mut Client) -> Result<()>
             info!("creating system extensions with query: {}", query);
             client.simple_query(query)?;
         }
-        if libs.contains("pg_wait_sampling") {
-            // Create extension only if this compute really needs it
-            let query = "CREATE EXTENSION IF NOT EXISTS pg_wait_sampling";
-            info!("creating system extensions with query: {}", query);
-            client.simple_query(query)?;
-        }
     }
 
     Ok(())

control_plane/src/bin/neon_local.rs

@@ -587,6 +587,31 @@ fn handle_timeline(timeline_match: &ArgMatches, env: &mut local_env::LocalEnv) -
                 timeline_info.timeline_id
             );
         }
+        Some(("merge", branch_match)) => {
+            let src_endpoint_id = branch_match
+                .get_one::<String>("src-endpoint")
+                .map(|s| s.as_str())
+                .ok_or(anyhow!("No source endpoint provided"))?;
+            let dst_endpoint_id = branch_match
+                .get_one::<String>("dst-endpoint")
+                .map(|s| s.as_str())
+                .ok_or(anyhow!("No destination endpoint provided"))?;
+
+            let cplane = ComputeControlPlane::load(env.clone())?;
+            let src_endpoint = cplane.endpoints.get(src_endpoint_id).unwrap();
+            let dst_endpoint = cplane.endpoints.get(dst_endpoint_id).unwrap();
+            dst_endpoint.merge_from(src_endpoint)?;
+        }
+        Some(("set_mergeable", branch_match)) => {
+            let endpoint_id = branch_match
+                .get_one::<String>("endpoint")
+                .map(|s| s.as_str())
+                .ok_or(anyhow!("No endpoint provided"))?;
+
+            let cplane = ComputeControlPlane::load(env.clone())?;
+            let endpoint = cplane.endpoints.get(endpoint_id).unwrap();
+            endpoint.set_mergeable()?;
+        }
         Some((sub_name, _)) => bail!("Unexpected tenant subcommand '{sub_name}'"),
         None => bail!("no tenant subcommand provided"),
     }
@@ -1305,6 +1330,15 @@ fn cli() -> Command {
                     .help("Use last Lsn of another timeline (and its data) as base when creating the new timeline. The timeline gets resolved by its branch name.").required(false))
                 .arg(Arg::new("ancestor-start-lsn").long("ancestor-start-lsn")
                     .help("When using another timeline as base, use a specific Lsn in it instead of the latest one").required(false)))
+            .subcommand(Command::new("merge")
+                        .about("Merge changes from one branch into another")
+                        .arg(Arg::new("src-endpoint").long("src-endpoint").help("Source endpoint for merge").required(true))
+                        .arg(Arg::new("dst-endpoint").long("dst-endpoint").help("Destination endpoint for merge").required(true))
+            )
+            .subcommand(Command::new("set_mergeable")
+                        .about("Mark branch as mergeable")
+                        .arg(Arg::new("endpoint").long("endpoint").help("Enpoint to be marked as mergeable").required(true))
+            )
             .subcommand(Command::new("create")
                 .about("Create a new blank timeline")
                 .arg(tenant_id_arg.clone())

control_plane/src/endpoint.rs

@@ -572,7 +572,11 @@ impl Endpoint {
                         }
                         ComputeStatus::Empty
                         | ComputeStatus::ConfigurationPending
-                        | ComputeStatus::Configuration => {
+                        | ComputeStatus::Configuration
+                        | ComputeStatus::MergePending
+                        | ComputeStatus::Merging
+                        | ComputeStatus::SetMergeablePending
+                        | ComputeStatus::SetMergeable => {
                             bail!("unexpected compute status: {:?}", state.status)
                         }
                     }
@@ -674,6 +678,51 @@ impl Endpoint {
         }
     }
 
+    pub fn merge_from(&self, merge_from: &Arc<Endpoint>) -> Result<()> {
+        let client = reqwest::blocking::Client::new();
+        let response = client
+            .post(format!(
+                "http://{}:{}/merge",
+                self.http_address.ip(),
+                self.http_address.port()
+            ))
+            .body(merge_from.connstr())
+            .send()?;
+        let status = response.status();
+        if !(status.is_client_error() || status.is_server_error()) {
+            Ok(())
+        } else {
+            let url = response.url().to_owned();
+            let msg = match response.text() {
+                Ok(err_body) => format!("Error: {}", err_body),
+                Err(_) => format!("Http error ({}) at {}.", status.as_u16(), url),
+            };
+            Err(anyhow::anyhow!(msg))
+        }
+    }
+
+    pub fn set_mergeable(&self) -> Result<()> {
+        let client = reqwest::blocking::Client::new();
+        let response = client
+            .post(format!(
+                "http://{}:{}/set_mergeable",
+                self.http_address.ip(),
+                self.http_address.port()
+            ))
+            .send()?;
+        let status = response.status();
+        if !(status.is_client_error() || status.is_server_error()) {
+            Ok(())
+        } else {
+            let url = response.url().to_owned();
+            let msg = match response.text() {
+                Ok(err_body) => format!("Error: {}", err_body),
+                Err(_) => format!("Http error ({}) at {}.", status.as_u16(), url),
+            };
+            Err(anyhow::anyhow!(msg))
+        }
+    }
+
     pub fn stop(&self, destroy: bool) -> Result<()> {
         // If we are going to destroy data directory,
         // use immediate shutdown mode, otherwise,

libs/compute_api/src/responses.rs

@@ -48,6 +48,14 @@ pub enum ComputeStatus {
     Running,
     // New spec is being applied.
     Configuration,
+    // Merge requested
+    MergePending,
+    // Set mergeable requested
+    SetMergeablePending,
+    // Merge in progress
+    Merging,
+    // Set mergeable in progress
+    SetMergeable,
     // Either startup or configuration failed,
     // compute will exit soon or is waiting for
     // control-plane to terminate it.

proxy/src/auth/credentials.rs

@@ -1,9 +1,7 @@
 //! User credentials used in authentication.
 
 use crate::{
-    auth::password_hack::parse_endpoint_param,
-    error::UserFacingError,
-    proxy::{neon_options, NUM_CONNECTION_ACCEPTED_BY_SNI},
+    auth::password_hack::parse_endpoint_param, error::UserFacingError, proxy::neon_options,
 };
 use itertools::Itertools;
 use pq_proto::StartupMessageParams;
@@ -126,22 +124,6 @@ impl<'a> ClientCredentials<'a> {
         .transpose()?;
 
         info!(user, project = project.as_deref(), "credentials");
-        if sni.is_some() {
-            info!("Connection with sni");
-            NUM_CONNECTION_ACCEPTED_BY_SNI
-                .with_label_values(&["sni"])
-                .inc();
-        } else if project.is_some() {
-            NUM_CONNECTION_ACCEPTED_BY_SNI
-                .with_label_values(&["no_sni"])
-                .inc();
-            info!("Connection without sni");
-        } else {
-            NUM_CONNECTION_ACCEPTED_BY_SNI
-                .with_label_values(&["password_hack"])
-                .inc();
-            info!("Connection with password hack");
-        }
 
         let cache_key = format!(
             "{}{}",

proxy/src/compute.rs

@@ -248,7 +248,6 @@ impl ConnCfg {
 
         // connect_raw() will not use TLS if sslmode is "disable"
         let (client, connection) = self.0.connect_raw(stream, tls).await?;
-        tracing::Span::current().record("pid", &tracing::field::display(client.get_process_id()));
         let stream = connection.stream.into_inner();
 
         info!(

proxy/src/proxy.rs

@@ -129,15 +129,6 @@ pub static RATE_LIMITER_LIMIT: Lazy<IntGaugeVec> = Lazy::new(|| {
     .unwrap()
 });
 
-pub static NUM_CONNECTION_ACCEPTED_BY_SNI: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        "proxy_accepted_connections_by_sni",
-        "Number of connections (per sni).",
-        &["kind"],
-    )
-    .unwrap()
-});
-
 pub struct LatencyTimer {
     // time since the stopwatch was started
     start: Option<Instant>,
@@ -514,7 +505,7 @@ pub fn invalidate_cache(node_info: console::CachedNodeInfo) -> compute::ConnCfg
 }
 
 /// Try to connect to the compute node once.
-#[tracing::instrument(name = "connect_once", fields(pid = tracing::field::Empty), skip_all)]
+#[tracing::instrument(name = "connect_once", skip_all)]
 async fn connect_to_compute_once(
     node_info: &console::CachedNodeInfo,
     timeout: time::Duration,

proxy/src/serverless/conn_pool.rs

@@ -208,13 +208,14 @@ impl GlobalConnPool {
             } else {
                 info!("pool: reusing connection '{conn_info}'");
                 client.session.send(session_id)?;
-                tracing::Span::current().record(
-                    "pid",
-                    &tracing::field::display(client.inner.get_process_id()),
-                );
                 latency_timer.pool_hit();
                 latency_timer.success();
-                return Ok(Client::new(client, pool).await);
+                return Ok(Client {
+                    conn_id: client.conn_id,
+                    inner: Some(client),
+                    span: Span::current(),
+                    pool,
+                });
             }
         } else {
             let conn_id = uuid::Uuid::new_v4();
@@ -228,12 +229,6 @@ impl GlobalConnPool {
             )
             .await
         };
-        if let Ok(client) = &new_client {
-            tracing::Span::current().record(
-                "pid",
-                &tracing::field::display(client.inner.get_process_id()),
-            );
-        }
 
         match &new_client {
             // clear the hash. it's no longer valid
@@ -267,8 +262,13 @@ impl GlobalConnPool {
             }
             _ => {}
         }
-        let new_client = new_client?;
-        Ok(Client::new(new_client, pool).await)
+
+        new_client.map(|inner| Client {
+            conn_id: inner.conn_id,
+            inner: Some(inner),
+            span: Span::current(),
+            pool,
+        })
     }
 
     fn put(&self, conn_info: &ConnInfo, client: ClientInner) -> anyhow::Result<()> {
@@ -394,7 +394,7 @@ impl ConnectMechanism for TokioMechanism<'_> {
 // Wake up the destination if needed. Code here is a bit involved because
 // we reuse the code from the usual proxy and we need to prepare few structures
 // that this code expects.
-#[tracing::instrument(fields(pid = tracing::field::Empty), skip_all)]
+#[tracing::instrument(skip_all)]
 async fn connect_to_compute(
     config: &config::ProxyConfig,
     conn_info: &ConnInfo,
@@ -461,7 +461,6 @@ async fn connect_to_compute_once(
         .connect_timeout(timeout)
         .connect(tokio_postgres::NoTls)
         .await?;
-    tracing::Span::current().record("pid", &tracing::field::display(client.get_process_id()));
 
     let (tx, mut rx) = tokio::sync::watch::channel(session);
 
@@ -548,17 +547,6 @@ pub struct Discard<'a> {
 }
 
 impl Client {
-    pub(self) async fn new(
-        inner: ClientInner,
-        pool: Option<(ConnInfo, Arc<GlobalConnPool>)>,
-    ) -> Self {
-        Self {
-            conn_id: inner.conn_id,
-            inner: Some(inner),
-            span: Span::current(),
-            pool,
-        }
-    }
     pub fn inner(&mut self) -> (&mut tokio_postgres::Client, Discard<'_>) {
         let Self {
             inner,

proxy/src/serverless/sql_over_http.rs

@@ -250,7 +250,7 @@ pub async fn handle(
     Ok(response)
 }
 
-#[instrument(name = "sql-over-http", fields(pid = tracing::field::Empty), skip_all)]
+#[instrument(name = "sql-over-http", skip_all)]
 async fn handle_inner(
     request: Request<Body>,
     sni_hostname: Option<String>,

test_runner/fixtures/neon_fixtures.py

@@ -1154,6 +1154,42 @@ class NeonCli(AbstractNeonCli):
         kwargs["local_binpath"] = True
         return super().raw_cli(*args, **kwargs)
 
+
+    def merge(
+        self,
+        src_endpoint: Endpoint,
+        dst_endpoint: Endpoint):
+        """
+        Merge two branches
+        """
+
+        args = [
+            "timeline",
+            "merge",
+            "--src-endpoint",
+            str(src_endpoint.endpoint_id),
+            "--dst-endpoint",
+            str(dst_endpoint.endpoint_id)
+        ]
+        res = self.raw_cli(args)
+        res.check_returncode()
+
+    def set_mergeable(
+        self,
+        endpoint: Endpoint):
+        """
+        Merge two branches
+        """
+
+        args = [
+            "timeline",
+            "set_mergeable",
+            "--endpoint",
+            str(endpoint.endpoint_id),
+        ]
+        res = self.raw_cli(args)
+        res.check_returncode()
+
     def create_tenant(
         self,
         tenant_id: Optional[TenantId] = None,

test_runner/regress/test_merge.py

@@ -0,0 +1,43 @@
+import time
+from fixtures.log_helper import log
+from fixtures.neon_fixtures import NeonEnvBuilder
+from fixtures.types import TimelineId
+from fixtures.utils import query_scalar
+
+#
+# Merge ancestor branch with the main branch.
+#
+def test_merge(neon_env_builder: NeonEnvBuilder):
+    env = neon_env_builder.init_start()
+    pageserver_http = env.pageserver.http_client()
+
+    # Override defaults: 4M checkpoint_distance, disable background compaction and gc.
+    tenant, _ = env.neon_cli.create_tenant()
+
+    main_branch = env.endpoints.create_start("main", tenant_id=tenant)
+    main_cur = main_branch.connect().cursor()
+
+    # Create table and insert some data
+    main_cur.execute("CREATE TABLE t(x bigint primary key)")
+    main_cur.execute("INSERT INTO t values(generate_series(1, 10000))");
+
+    # Create branch ws.
+    env.neon_cli.create_branch("ws", "main", tenant_id=tenant)
+    ws_branch = env.endpoints.create_start("ws", tenant_id=tenant)
+    log.info("postgres is running on 'ws' branch")
+
+    # Merge brnach ws as mergeable:it create logical replication slots and pins WAL
+    env.neon_cli.set_mergeable(ws_branch)
+
+    # Insert more data in the branch
+    ws_cur = ws_branch.connect().cursor()
+    ws_cur.execute("INSERT INTO t values(generate_series(10001, 20000))")
+
+    # Merge ws brnach intp main
+    env.neon_cli.merge(ws_branch, main_branch)
+
+    # sleep for some time until changes are applied
+    time.sleep(2)
+
+    # Check that changes are merged
+    assert query_scalar(main_cur, "SELECT count(*) from t") == 20000

test_runner/regress/test_normal_work.py

@@ -6,14 +6,12 @@ from fixtures.pageserver.http import PageserverHttpClient
 
 def check_tenant(env: NeonEnv, pageserver_http: PageserverHttpClient):
     tenant_id, timeline_id = env.neon_cli.create_tenant()
-    endpoint = env.endpoints.create_start("main", tenant_id=tenant_id, config_lines=[
-                "log_statement=all",
-            ],)
+    endpoint = env.endpoints.create_start("main", tenant_id=tenant_id)
     # we rely upon autocommit after each statement
     res_1 = endpoint.safe_psql_many(
         queries=[
-            "CREATE TABLE \nt(key int primary key, value text)",
-            "INSERT INTO \n\nt SELECT generate_series(1,100000), 'payload'",
+            "CREATE TABLE t(key int primary key, value text)",
+            "INSERT INTO t SELECT generate_series(1,100000), 'payload'",
             "SELECT sum(key) FROM t",
         ]
     )

vm-image-spec.yaml

@@ -1,152 +0,0 @@
-# Supplemental file for neondatabase/autoscaling's vm-builder, for producing the VM compute image.
----
-commands:
-  - name: cgconfigparser
-    user: root
-    sysvInitAction: sysinit
-    shell: "cgconfigparser -l /etc/cgconfig.conf -s 1664"
-  - name: pgbouncer
-    user: nobody
-    sysvInitAction: respawn
-    shell: "/usr/local/bin/pgbouncer /etc/pgbouncer.ini"
-  - name: postgres-exporter
-    user: nobody
-    sysvInitAction: respawn
-    shell: 'DATA_SOURCE_NAME="user=cloud_admin sslmode=disable dbname=postgres" /bin/postgres_exporter --extend.query-path /etc/postgres_exporter_queries.yml'
-shutdownHook: |
-  su -p postgres --session-command '/usr/local/bin/pg_ctl stop -D /var/db/postgres/compute/pgdata -m fast --wait -t 10'
-files:
-  - filename: pgbouncer.ini
-    content: |
-      [databases]
-      *=host=localhost port=5432 auth_user=cloud_admin
-      [pgbouncer]
-      listen_port=6432
-      listen_addr=0.0.0.0
-      auth_type=scram-sha-256
-      auth_user=cloud_admin
-      auth_dbname=postgres
-      client_tls_sslmode=disable
-      server_tls_sslmode=disable
-      pool_mode=transaction
-      max_client_conn=10000
-      default_pool_size=16
-      max_prepared_statements=0
-  - filename: cgconfig.conf
-    content: |
-      # Configuration for cgroups in VM compute nodes
-      group neon-postgres {
-          perm {
-              admin {
-                  uid = postgres;
-              }
-              task {
-                  gid = users;
-              }
-          }
-          memory {}
-      }
-  - filename: postgres_exporter_queries.yml
-    content: |
-      pg_wait_sampling:
-        query: "select pid, event_type, event, w.queryid as queryid, query, count from pg_wait_sampling_profile w left join pg_stat_statements s on w.queryid = s.queryid;"
-        cache_seconds: 30
-        metrics:
-          - pid:
-              usage: "LABEL"
-              description: "backend pid"
-          - event_type:
-              usage: "LABEL"
-              description: "event type"
-          - event:
-              usage: "LABEL"
-              description: "event"
-          - queryid:
-              usage: "LABEL"
-              description: "queryid"
-          - query:
-              usage: "LABEL"
-              description: "query"
-          - count:
-              usage: "GAUGE"
-              description: "count"
-build: |
-  # Build cgroup-tools
-  #
-  # At time of writing (2023-03-14), debian bullseye has a version of cgroup-tools (technically
-  # libcgroup) that doesn't support cgroup v2 (version 0.41-11). Unfortunately, the vm-monitor
-  # requires cgroup v2, so we'll build cgroup-tools ourselves.
-  FROM debian:bullseye-slim as libcgroup-builder
-  ENV LIBCGROUP_VERSION v2.0.3
-
-  RUN set -exu \
-      && apt update \
-      && apt install --no-install-recommends -y \
-          git \
-          ca-certificates \
-          automake \
-          cmake \
-          make \
-          gcc \
-          byacc \
-          flex \
-          libtool \
-          libpam0g-dev \
-      && git clone --depth 1 -b $LIBCGROUP_VERSION https://github.com/libcgroup/libcgroup \
-      && INSTALL_DIR="/libcgroup-install" \
-      && mkdir -p "$INSTALL_DIR/bin" "$INSTALL_DIR/include" \
-      && cd libcgroup \
-      # extracted from bootstrap.sh, with modified flags:
-      && (test -d m4 || mkdir m4) \
-      && autoreconf -fi \
-      && rm -rf autom4te.cache \
-      && CFLAGS="-O3" ./configure --prefix="$INSTALL_DIR" --sysconfdir=/etc --localstatedir=/var --enable-opaque-hierarchy="name=systemd" \
-      # actually build the thing...
-      && make install
-
-  FROM quay.io/prometheuscommunity/postgres-exporter:v0.12.0 AS postgres-exporter
-
-  # Build pgbouncer
-  #
-  FROM debian:bullseye-slim AS pgbouncer
-  RUN set -e \
-      && apt-get update \
-      && apt-get install -y \
-          curl \
-          build-essential \
-          pkg-config \
-          libevent-dev \
-          libssl-dev
-
-  ENV PGBOUNCER_VERSION 1.21.0
-  ENV PGBOUNCER_GITPATH 1_21_0
-  RUN set -e \
-      && curl -sfSL https://github.com/pgbouncer/pgbouncer/releases/download/pgbouncer_${PGBOUNCER_GITPATH}/pgbouncer-${PGBOUNCER_VERSION}.tar.gz -o pgbouncer-${PGBOUNCER_VERSION}.tar.gz \
-      && tar xzvf pgbouncer-${PGBOUNCER_VERSION}.tar.gz \
-      && cd pgbouncer-${PGBOUNCER_VERSION} \
-      && LDFLAGS=-static ./configure --prefix=/usr/local/pgbouncer --without-openssl \
-      && make -j $(nproc) \
-      && make install
-merge: |
-  # tweak nofile limits
-  RUN set -e \
-      && echo 'fs.file-max = 1048576' >>/etc/sysctl.conf \
-      && test ! -e /etc/security || ( \
-         echo '*    - nofile 1048576' >>/etc/security/limits.conf \
-      && echo 'root - nofile 1048576' >>/etc/security/limits.conf \
-         )
-
-  COPY cgconfig.conf /etc/cgconfig.conf
-  COPY pgbouncer.ini /etc/pgbouncer.ini
-  COPY postgres_exporter_queries.yml /etc/postgres_exporter_queries.yml
-  RUN set -e \
-      && chown postgres:postgres /etc/pgbouncer.ini \
-      && chmod 0644 /etc/pgbouncer.ini \
-      && chmod 0644 /etc/cgconfig.conf \
-      && chmod 0644 /etc/postgres_exporter_queries.yml
-
-  COPY --from=libcgroup-builder /libcgroup-install/bin/*  /usr/bin/
-  COPY --from=libcgroup-builder /libcgroup-install/lib/*  /usr/lib/
-  COPY --from=libcgroup-builder /libcgroup-install/sbin/* /usr/sbin/
-  COPY --from=postgres-exporter /bin/postgres_exporter /bin/postgres_exporter
-  COPY --from=pgbouncer         /usr/local/pgbouncer/bin/pgbouncer /usr/local/bin/pgbouncer