Not working version

.circleci/ansible/.gitignore

@@ -1,4 +1,2 @@
 zenith_install.tar.gz
 .zenith_current_version
-neon_install.tar.gz
-.neon_current_version

.circleci/config.yml

@@ -121,7 +121,7 @@ jobs:
             export RUSTC_WRAPPER=cachepot
             export AWS_ACCESS_KEY_ID="${CACHEPOT_AWS_ACCESS_KEY_ID}"
             export AWS_SECRET_ACCESS_KEY="${CACHEPOT_AWS_SECRET_ACCESS_KEY}"
-            "${cov_prefix[@]}" mold -run cargo build $CARGO_FLAGS --features failpoints --bins --tests
+            "${cov_prefix[@]}" mold -run cargo build $CARGO_FLAGS --bins --tests
             cachepot -s
 
       - save_cache:
@@ -579,13 +579,13 @@ jobs:
           name: Setup helm v3
           command: |
             curl -s https://raw.githubusercontent.com/helm/helm/main/scripts/get-helm-3 | bash
-            helm repo add neondatabase https://neondatabase.github.io/helm-charts
+            helm repo add zenithdb https://neondatabase.github.io/helm-charts
       - run:
           name: Re-deploy proxy
           command: |
             DOCKER_TAG=$(git log --oneline|wc -l)
-            helm upgrade zenith-proxy     neondatabase/neon-proxy --install -f .circleci/helm-values/staging.proxy.yaml --set image.tag=${DOCKER_TAG} --wait
-            helm upgrade neon-proxy-scram neondatabase/neon-proxy --install -f .circleci/helm-values/staging.proxy-scram.yaml --set image.tag=${DOCKER_TAG} --wait
+            helm upgrade zenith-proxy zenithdb/zenith-proxy --install -f .circleci/helm-values/staging.proxy.yaml --set image.tag=${DOCKER_TAG} --wait
+
 
   deploy-release:
     docker:

.circleci/helm-values/staging.proxy-scram.yaml

@@ -1,30 +0,0 @@
-# Helm chart values for zenith-proxy.
-# This is a YAML-formatted file.
-
-image:
-  repository: neondatabase/neon
-
-settings:
-  authBackend: "console"
-  authEndpoint: "http://console-staging.local:9095/management/api/v2"
-
-# -- Additional labels for zenith-proxy pods
-podLabels:
-  zenith_service: proxy-scram
-  zenith_env: staging
-  zenith_region: us-east-1
-  zenith_region_slug: virginia
-
-exposedService:
-  annotations:
-    service.beta.kubernetes.io/aws-load-balancer-type: external
-    service.beta.kubernetes.io/aws-load-balancer-nlb-target-type: ip
-    service.beta.kubernetes.io/aws-load-balancer-scheme: internet-facing
-    external-dns.alpha.kubernetes.io/hostname: *.cloud.stage.neon.tech
-
-metrics:
-  enabled: true
-  serviceMonitor:
-    enabled: true
-    selector:
-      release: kube-prometheus-stack

.gitignore

@@ -11,6 +11,3 @@ test_output/
 # Coverage
 *.profraw
 *.profdata
-
-*.key
-*.crt

Cargo.lock

@@ -249,6 +249,9 @@ name = "cc"
 version = "1.0.72"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "22a9137b95ea06864e018375b72adfb7db6e6f68cfc8df5a04d00288050485ee"
+dependencies = [
+ "jobserver",
+]
 
 [[package]]
 name = "cexpr"
@@ -1226,6 +1229,15 @@ version = "1.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1aab8fc367588b89dcee83ab0fd66b72b50b72fa1904d7095045ace2b0c81c35"
 
+[[package]]
+name = "jobserver"
+version = "0.1.24"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "af25a77299a7f711a01975c35a6a424eb6862092cc2d6c72c4ed6cbc56dfc1fa"
+dependencies = [
+ "libc",
+]
+
 [[package]]
 name = "js-sys"
 version = "0.3.56"
@@ -1669,6 +1681,7 @@ dependencies = [
  "url",
  "utils",
  "workspace_hack",
+ "zstd",
 ]
 
 [[package]]
@@ -2040,7 +2053,6 @@ dependencies = [
  "tokio-postgres",
  "tokio-postgres-rustls",
  "tokio-rustls",
- "url",
  "utils",
  "workspace_hack",
 ]
@@ -3657,3 +3669,32 @@ name = "zeroize"
 version = "1.5.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7c88870063c39ee00ec285a2f8d6a966e5b6fb2becc4e8dac77ed0d370ed6006"
+
+[[package]]
+name = "zstd"
+version = "0.11.1+zstd.1.5.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "77a16b8414fde0414e90c612eba70985577451c4c504b99885ebed24762cb81a"
+dependencies = [
+ "zstd-safe",
+]
+
+[[package]]
+name = "zstd-safe"
+version = "5.0.1+zstd.1.5.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7c12659121420dd6365c5c3de4901f97145b79651fb1d25814020ed2ed0585ae"
+dependencies = [
+ "libc",
+ "zstd-sys",
+]
+
+[[package]]
+name = "zstd-sys"
+version = "2.0.1+zstd.1.5.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9fd07cbbc53846d9145dbffdf6dd09a7a0aa52be46741825f5c97bdd4f73f12b"
+dependencies = [
+ "cc",
+ "libc",
+]

docs/rfcs/016-connection-routing.md

@@ -1,151 +0,0 @@
-# Dispatching a connection
-
-For each client connection, Neon service needs to authenticate the
-connection, and route it to the right PostgreSQL instance.
-
-## Authentication
-
-There are three different ways to authenticate:
-
-- anonymous; no authentication needed
-- PostgreSQL authentication
-- github single sign-on using browser
-
-In anonymous access, the user doesn't need to perform any
-authentication at all. This can be used e.g. in interactive PostgreSQL
-documentation, allowing you to run the examples very quickly. Similar
-to sqlfiddle.com.
-
-PostgreSQL authentication works the same as always. All the different
-PostgreSQL authentication options like SCRAM, kerberos, etc. are
-available. [1]
-
-The third option is to authenticate with github single sign-on. When
-you open the connection in psql, you get a link that you open with
-your browser. Opening the link redirects you to github authentication,
-and lets the connection to proceed. This is also known as "Link auth" [2].
-
-
-## Routing the connection
-
-When a client starts a connection, it needs to be routed to the
-correct PostgreSQL instance. Routing can be done by the proxy, acting
-as a man-in-the-middle, or the connection can be routed at the network
-level based on the hostname or IP address.
-
-Either way, Neon needs to identify which PostgreSQL instance the
-connection should be routed to. If the instance is not already
-running, it needs to be started. Some connections always require a new
-PostgreSQL instance to be created, e.g. if you want to run a one-off
-query against a particular point-in-time.
-
-The PostgreSQL instance is identified by:
-- Neon account (possibly anonymous)
-- cluster (known as tenant in the storage?)
-- branch or snapshot name
-- timestamp (PITR)
-- primary or read-replica
-- one-off read replica
-- one-off writeable branch
-
-When you are using regular PostgreSQL authentication or anonymous
-access, the connection URL needs to contain all the information needed
-for the routing. With github single sign-on, the browser is involved
-and some details - the Neon account in particular - can be deduced
-from the authentication exchange.
-
-There are three methods for identifying the PostgreSQL instance:
-
-- Browser interaction (link auth)
-- Options in the connection URL and the domain name
-- A pre-defined endpoint, identified by domain name or IP address
-
-### Link Auth
-
-    postgres://<username>@start.neon.tech/<dbname>
-
-This gives you a link that you open in browser. Clicking the link
-performs github authentication, and the Neon account name is
-provided to the proxy behind the scenes. The proxy routes the
-connection to the primary PostgreSQL instance in cluster called
-"main", branch "main".
-
-Further ideas:
-- You could pre-define a different target for link auth
-  connections in the UI.
-- You could have a drop-down in the browser, allowing you to connect
-  to any cluster you want. Link Auth can be like Teleport.
-
-### Connection URL
-
-The connection URL looks like this:
-
-    postgres://<username>@<cluster-id>.db.neon.tech/<dbname>
-
-By default, this connects you to the primary PostgreSQL instance
-running on the "main" branch in the named cluster [3]. However, you can
-change that by specifying options in the connection URL. The following
-options are supported:
-
-| option name  | Description                                                                                       | Examples                                            |
-| ---          | ---                                                                                               | ---                                                 |
-| cluster      | Cluster name                                                                                      | cluster:myproject                                   |
-| branch       | Branch name                                                                                       | branch:main                                         |
-| timestamp    | Connect to an instance at given point-in-time.                                                    | timestamp:2022-04-08 timestamp:2022-04-08T11:42:16Z |
-| lsn          | Connect to an instance at given LSN                                                               | lsn:0/12FF0420                                      |
-| read-replica | Connect to a read-replica. If the parameter is 'new', a new instance is created for this session. | read-replica read-replica:new                       |
-
-For example, to read branch 'testing' as it was on Mar 31, 2022, you could
-specify a timestamp in the connection URL [4]:
-
-    postgres://alice@cluster-1234.db.neon.tech/postgres?options=branch:testing,timestamp:2022-03-31
-
-Connecting with cluster name and options can be disabled in the UI. If
-disabled, you can only connect using a pre-defined endpoint.
-
-### Pre-defined Endpoint
-
-Instead of providing the cluster name, branch, and all those options
-in the connection URL, you can define a named endpoint with the same
-options.
-
-In the UI, click "create endpoint". Fill in the details:
-
-- Cluster name
-- Branch
-- timestamp or LSN
-- is this for the primary or for a read replica
-- etc.
-
-When you click Finish, a named endpoint is created. You can now use the endpoint ID to connect:
-
-    postgres://<username>@<endpoint-id>.endpoint.neon.tech/<dbname>
-
-
-An endpoint can be assigned a static or dynamic IP address, so that
-you can connect to it with clients that don't support TLS SNI. Maybe
-bypass the proxy altogether, but that ought to be invisible to the
-user.
-
-You can limit the range of source IP addresses that are allowed to
-connect to an endpoint. An endpoint can also be exposed in an Amazon
-VPC, allowing direct connections from applications.
-
-
-# Footnotes
-
-[1] I'm not sure how feasible it is to set up configure like Kerberos
-or LDAP in a cloud environment. But in principle I think we should
-allow customers to have the full power of PostgreSQL, including all
-authentication options. However, it's up to the customer to configure
-it correctly.
-
-[2] Link is a way to both authenticate and to route the connection
-
-[3] This assumes that cluster-ids are globally unique, across all
-Neon accounts.
-
-[4] The syntax accepted in the connection URL is limited by libpq. The
-only way to pass arbitrary options to the server (or our proxy) is
-with the "options" keyword, and the options must be percent-encoded. I
-think the above would work but i haven't tested it

libs/postgres_ffi/src/xlog_utils.rs

@@ -118,15 +118,11 @@ pub fn normalize_lsn(lsn: Lsn, seg_sz: usize) -> Lsn {
 }
 
 pub fn get_current_timestamp() -> TimestampTz {
-    to_pg_timestamp(SystemTime::now())
-}
-
-pub fn to_pg_timestamp(time: SystemTime) -> TimestampTz {
     const UNIX_EPOCH_JDATE: u64 = 2440588; /* == date2j(1970, 1, 1) */
     const POSTGRES_EPOCH_JDATE: u64 = 2451545; /* == date2j(2000, 1, 1) */
     const SECS_PER_DAY: u64 = 86400;
     const USECS_PER_SEC: u64 = 1000000;
-    match time.duration_since(SystemTime::UNIX_EPOCH) {
+    match SystemTime::now().duration_since(SystemTime::UNIX_EPOCH) {
         Ok(n) => {
             ((n.as_secs() - ((POSTGRES_EPOCH_JDATE - UNIX_EPOCH_JDATE) * SECS_PER_DAY))
                 * USECS_PER_SEC

libs/utils/src/pq_proto.rs

@@ -503,18 +503,6 @@ impl RowDescriptor<'_> {
             formatcode: 0,
         }
     }
-
-    pub const fn text_col(name: &[u8]) -> RowDescriptor {
-        RowDescriptor {
-            name,
-            tableoid: 0,
-            attnum: 0,
-            typoid: TEXT_OID,
-            typlen: -1,
-            typmod: 0,
-            formatcode: 0,
-        }
-    }
 }
 
 #[derive(Debug)]

pageserver/Cargo.toml

@@ -58,6 +58,8 @@ rusoto_core = "0.47"
 rusoto_s3 = "0.47"
 async-trait = "0.1"
 
+zstd = "0.11.1"
+
 postgres_ffi = { path = "../libs/postgres_ffi" }
 metrics = { path = "../libs/metrics" }
 utils = { path = "../libs/utils" }

pageserver/src/basebackup.rs

@@ -154,17 +154,9 @@ impl<'a> Basebackup<'a> {
             let img = self
                 .timeline
                 .get_slru_page_at_lsn(slru, segno, blknum, self.lsn)?;
+            ensure!(img.len() == pg_constants::BLCKSZ as usize);
 
-            if slru == SlruKind::Clog {
-                ensure!(
-                    img.len() == pg_constants::BLCKSZ as usize
-                        || img.len() == pg_constants::BLCKSZ as usize + 8
-                );
-            } else {
-                ensure!(img.len() == pg_constants::BLCKSZ as usize);
-            }
-
-            slru_buf.extend_from_slice(&img[..pg_constants::BLCKSZ as usize]);
+            slru_buf.extend_from_slice(&img);
         }
 
         let segname = format!("{}/{:>04X}", slru.to_str(), segno);

pageserver/src/bin/pageserver.rs

@@ -8,7 +8,6 @@ use anyhow::{bail, Context, Result};
 use clap::{App, Arg};
 use daemonize::Daemonize;
 
-use fail::FailScenario;
 use pageserver::{
     config::{defaults::*, PageServerConf},
     http, page_cache, page_service, profiling, tenant_mgr, thread_mgr,
@@ -85,23 +84,8 @@ fn main() -> anyhow::Result<()> {
                 .help("Additional configuration overrides of the ones from the toml config file (or new ones to add there).
                 Any option has to be a valid toml document, example: `-c=\"foo='hey'\"` `-c=\"foo={value=1}\"`"),
         )
-        .arg(
-            Arg::new("enabled-features")
-                .long("enabled-features")
-                .takes_value(false)
-                .help("Show enabled compile time features"),
-        )
         .get_matches();
 
-    if arg_matches.is_present("enabled-features") {
-        let features: &[&str] = &[
-            #[cfg(feature = "failpoints")]
-            "failpoints",
-        ];
-        println!("{{\"features\": {features:?} }}");
-        return Ok(());
-    }
-
     let workdir = Path::new(arg_matches.value_of("workdir").unwrap_or(".zenith"));
     let workdir = workdir
         .canonicalize()
@@ -182,14 +166,6 @@ fn main() -> anyhow::Result<()> {
     // as a ref.
     let conf: &'static PageServerConf = Box::leak(Box::new(conf));
 
-    // If failpoints are used, terminate the whole pageserver process if they are hit.
-    let scenario = FailScenario::setup();
-    if fail::has_failpoints() {
-        std::panic::set_hook(Box::new(|_| {
-            std::process::exit(1);
-        }));
-    }
-
     // Basic initialization of things that don't change after startup
     virtual_file::init(conf.max_file_descriptors);
     page_cache::init(conf.page_cache_size);
@@ -205,12 +181,10 @@ fn main() -> anyhow::Result<()> {
                 cfg_file_path.display()
             )
         })?;
+        Ok(())
     } else {
-        start_pageserver(conf, daemonize).context("Failed to start pageserver")?;
+        start_pageserver(conf, daemonize).context("Failed to start pageserver")
     }
-
-    scenario.teardown();
-    Ok(())
 }
 
 fn start_pageserver(conf: &'static PageServerConf, daemonize: bool) -> Result<()> {
@@ -321,7 +295,7 @@ fn start_pageserver(conf: &'static PageServerConf, daemonize: bool) -> Result<()
                 signal.name()
             );
             profiling::exit_profiler(conf, &profiler_guard);
-            pageserver::shutdown_pageserver(0);
+            pageserver::shutdown_pageserver();
             unreachable!()
         }
     })

pageserver/src/bin/replay.rs

@@ -1,75 +0,0 @@
-use std::str::FromStr;
-
-use anyhow::Result;
-use postgres_ffi::{pg_constants::WAL_SEGMENT_SIZE, waldecoder::WalStreamDecoder};
-use utils::zid::{ZTenantId, ZTimelineId};
-use tokio::net::TcpStream;
-use utils::lsn::Lsn;
-
-
-struct PageServiceApi {
-    stream: TcpStream,
-}
-
-impl PageServiceApi {
-    async fn connect(tenant: &ZTenantId, timeline: &ZTimelineId, connstr: &str) -> Result<Self> {
-        let mut stream = TcpStream::connect("localhost:15000").await?;
-
-        // Connect to pageserver
-        // TODO read host, port, dbname, user from command line
-        let (client, conn) = tokio_postgres::Config::new()
-            .host("127.0.0.1")
-            .port(15000)
-            .dbname("postgres")
-            .user("zenith_admin")
-            .connect_raw(&mut stream, tokio_postgres::NoTls)
-            .await?;
-
-        let init_query = format!("callmemaybe {} {} {}", tenant, timeline, connstr);
-        tokio::select! {
-            _ = conn => panic!("connection closed during callmemaybe"),
-            _ = client.query(init_query.as_str(), &[]) => (),
-        };
-
-        Ok(Self { stream })
-    }
-}
-
-
-#[tokio::main]
-async fn main() -> Result<()> {
-    use clap::{App, Arg};
-    let arg_matches = App::new("Replay")
-        .arg(
-            Arg::new("tenant")
-                .long("tenant")
-                .takes_value(true)
-        )
-        .arg(
-            Arg::new("timeline")
-                .long("timeline")
-                .takes_value(true)
-        )
-        .get_matches();
-
-    let partial_path = "/home/bojan/tmp/sk_wal";
-    let startpos = Lsn(23761464);  // I got this by grepping sk log for "restart decoder"
-    let xlogoff: usize = startpos.segment_offset(WAL_SEGMENT_SIZE);
-
-    let mut decoder = WalStreamDecoder::new(startpos);
-    let bytes = std::fs::read(partial_path)?;
-    decoder.feed_bytes(&bytes[xlogoff..(xlogoff+10000)]);
-
-    while let Some((lsn, rec)) = decoder.poll_decode()? {
-        println!("lsn: {}", lsn);
-    }
-
-    // TODO start replication server, get connstr
-
-    let tenant = ZTenantId::from_str(arg_matches.value_of("tenant").unwrap())?;
-    let timeline = ZTimelineId::from_str(arg_matches.value_of("timeline").unwrap())?;
-    let connstr = "lol";
-    let mut api = PageServiceApi::connect(&tenant, &timeline, connstr).await?;
-
-    Ok(())
-}

pageserver/src/config.rs

@@ -20,6 +20,12 @@ use utils::{
 use crate::layered_repository::TIMELINES_SEGMENT_NAME;
 use crate::tenant_config::{TenantConf, TenantConfOpt};
 
+pub const ZSTD_MAX_SAMPLES: usize = 1024;
+pub const ZSTD_MIN_SAMPLES: usize = 8; // magic requirement of zstd
+pub const ZSTD_MAX_DICTIONARY_SIZE: usize = 64 * 1024;
+pub const ZSTD_COMPRESSION_LEVEL: i32 = 0; // default compression level
+pub const ZSTD_DECOMPRESS_BUFFER_LIMIT: usize = 64 * 1024; // TODO: handle larger WAL records?
+
 pub mod defaults {
     use crate::tenant_config::defaults::*;
     use const_format::formatcp;

pageserver/src/layered_repository.rs

@@ -17,6 +17,7 @@ use fail::fail_point;
 use itertools::Itertools;
 use lazy_static::lazy_static;
 use tracing::*;
+use utils::bin_ser::BeSer;
 
 use std::cmp::{max, min, Ordering};
 use std::collections::hash_map::Entry;
@@ -32,6 +33,7 @@ use std::sync::{Arc, Mutex, MutexGuard, RwLock, RwLockReadGuard, TryLockError};
 use std::time::{Duration, Instant, SystemTime};
 
 use self::metadata::{metadata_path, TimelineMetadata, METADATA_FILE_NAME};
+use crate::config;
 use crate::config::PageServerConf;
 use crate::keyspace::KeySpace;
 use crate::tenant_config::{TenantConf, TenantConfOpt};
@@ -1432,8 +1434,7 @@ impl LayeredTimeline {
 
             let layers = timeline.layers.read().unwrap();
 
-            // Check the open and frozen in-memory layers first, in order from newest
-            // to oldest.
+            // Check the open and frozen in-memory layers first
             if let Some(open_layer) = &layers.open_layer {
                 let start_lsn = open_layer.get_lsn_range().start;
                 if cont_lsn > start_lsn {
@@ -1451,7 +1452,7 @@ impl LayeredTimeline {
                     continue;
                 }
             }
-            for frozen_layer in layers.frozen_layers.iter().rev() {
+            for frozen_layer in layers.frozen_layers.iter() {
                 let start_lsn = frozen_layer.get_lsn_range().start;
                 if cont_lsn > start_lsn {
                     //info!("CHECKING for {} at {} on frozen layer {}", key, cont_lsn, frozen_layer.filename().display());
@@ -1504,20 +1505,12 @@ impl LayeredTimeline {
         let ancestor = self
             .ancestor_timeline
             .as_ref()
-            .with_context(|| {
-                format!(
-                    "Ancestor is missing. Timeline id: {} Ancestor id {:?}",
-                    self.timelineid,
-                    self.get_ancestor_timeline_id(),
-                )
-            })?
+            .expect("there should be an ancestor")
             .ensure_loaded()
             .with_context(|| {
                 format!(
-                    "Ancestor timeline is not is not loaded. Timeline id: {} Ancestor id {:?}",
-                    self.timelineid,
-                    self.get_ancestor_timeline_id(),
-                )
+                "Cannot get the whole layer for read locked: timeline {} is not present locally",
+                self.get_ancestor_timeline_id().unwrap())
             })?;
         Ok(Arc::clone(ancestor))
     }
@@ -1703,11 +1696,8 @@ impl LayeredTimeline {
             new_delta_path.clone(),
             self.conf.timeline_path(&self.timelineid, &self.tenantid),
         ])?;
-        fail_point!("checkpoint-before-sync");
 
-        fail_point!("flush-frozen");
-
-        // Finally, replace the frozen in-memory layer with the new on-disk layer
+        // Finally, replace the frozen in-memory layer with the new on-disk layers
         {
             let mut layers = self.layers.write().unwrap();
             let l = layers.frozen_layers.pop_front();
@@ -1728,7 +1718,6 @@ impl LayeredTimeline {
         // TODO: This perhaps should be done in 'flush_frozen_layers', after flushing
         // *all* the layers, to avoid fsyncing the file multiple times.
         let disk_consistent_lsn = Lsn(frozen_layer.get_lsn_range().end.0 - 1);
-        fail_point!("checkpoint-after-sync");
 
         // If we were able to advance 'disk_consistent_lsn', save it the metadata file.
         // After crash, we will restart WAL streaming and processing from that point.
@@ -1951,78 +1940,109 @@ impl LayeredTimeline {
             .map(|l| l.get_lsn_range())
             .reduce(|a, b| min(a.start, b.start)..max(a.end, b.end))
             .unwrap();
-
-        let all_values_iter = level0_deltas.iter().map(|l| l.iter()).kmerge_by(|a, b| {
-            if let Ok((a_key, a_lsn, _)) = a {
-                if let Ok((b_key, b_lsn, _)) = b {
-                    match a_key.cmp(b_key) {
-                        Ordering::Less => true,
-                        Ordering::Equal => a_lsn <= b_lsn,
-                        Ordering::Greater => false,
+        let mut new_layers = Vec::new();
+        {
+            let mut all_values_iter = level0_deltas.iter().map(|l| l.iter()).kmerge_by(|a, b| {
+                if let Ok((a_key, a_lsn, _)) = a {
+                    if let Ok((b_key, b_lsn, _)) = b {
+                        match a_key.cmp(b_key) {
+                            Ordering::Less => true,
+                            Ordering::Equal => a_lsn <= b_lsn,
+                            Ordering::Greater => false,
+                        }
+                    } else {
+                        false
                     }
                 } else {
-                    false
+                    true
                 }
-            } else {
-                true
-            }
-        });
+            });
 
-        // Merge the contents of all the input delta layers into a new set
-        // of delta layers, based on the current partitioning.
-        //
-        // TODO: this actually divides the layers into fixed-size chunks, not
-        // based on the partitioning.
-        //
-        // TODO: we should also opportunistically materialize and
-        // garbage collect what we can.
-        let mut new_layers = Vec::new();
-        let mut prev_key: Option<Key> = None;
-        let mut writer: Option<DeltaLayerWriter> = None;
-        for x in all_values_iter {
-            let (key, lsn, value) = x?;
+            // Merge the contents of all the input delta layers into a new set
+            // of delta layers, based on the current partitioning.
+            //
+            // TODO: this actually divides the layers into fixed-size chunks, not
+            // based on the partitioning.
+            //
+            // TODO: we should also opportunistically materialize and
+            // garbage collect what we can.
+            let mut prev_key: Option<Key> = None;
+            let mut writer: Option<DeltaLayerWriter> = None;
 
-            if let Some(prev_key) = prev_key {
-                if key != prev_key && writer.is_some() {
-                    let size = writer.as_mut().unwrap().size();
-                    if size > target_file_size {
-                        new_layers.push(writer.take().unwrap().finish(prev_key.next())?);
-                        writer = None;
+            while let Some(x) = all_values_iter.next() {
+                let (key, lsn, value) = x?;
+
+                if let Some(prev_key) = prev_key {
+                    if key != prev_key && writer.is_some() {
+                        let size = writer.as_mut().unwrap().size();
+                        if size > target_file_size {
+                            new_layers.push(writer.take().unwrap().finish(prev_key.next())?);
+                            writer = None;
+                        }
                     }
                 }
+
+                if writer.is_none() {
+                    let mut samples: Vec<Vec<u8>> = Vec::with_capacity(config::ZSTD_MAX_SAMPLES);
+                    let mut prefetched = Vec::with_capacity(config::ZSTD_MAX_SAMPLES);
+                    samples.push(Value::ser(&value)?);
+                    prefetched.push((key, lsn, value));
+
+                    while let Some(y) = all_values_iter.next() {
+                        if let Ok((key, lsn, value)) = y {
+                            samples.push(Value::ser(&value)?);
+                            prefetched.push((key, lsn, value));
+                            if samples.len() == config::ZSTD_MAX_SAMPLES {
+                                break;
+                            }
+                        } else {
+                            break;
+                        }
+                    }
+
+                    let dictionary = if samples.len() >= config::ZSTD_MIN_SAMPLES {
+                        zstd::dict::from_samples(&samples, config::ZSTD_MAX_DICTIONARY_SIZE)?
+                    } else {
+                        Vec::new()
+                    };
+
+                    writer = Some(DeltaLayerWriter::new(
+                        self.conf,
+                        self.timelineid,
+                        self.tenantid,
+                        key,
+                        lsn_range.clone(),
+                        dictionary,
+                    )?);
+
+                    // Number of sample is relatively small, so we should not exceed target file limit
+                    for (key, lsn, value) in prefetched {
+                        writer.as_mut().unwrap().put_value(key, lsn, value)?;
+                        prev_key = Some(key);
+                    }
+                } else {
+                    writer.as_mut().unwrap().put_value(key, lsn, value)?;
+                    prev_key = Some(key);
+                }
+            }
+            if let Some(writer) = writer {
+                new_layers.push(writer.finish(prev_key.unwrap().next())?);
             }
 
-            if writer.is_none() {
-                writer = Some(DeltaLayerWriter::new(
-                    self.conf,
-                    self.timelineid,
-                    self.tenantid,
-                    key,
-                    lsn_range.clone(),
-                )?);
+            // Sync layers
+            if !new_layers.is_empty() {
+                let mut layer_paths: Vec<PathBuf> = new_layers.iter().map(|l| l.path()).collect();
+
+                // also sync the directory
+                layer_paths.push(self.conf.timeline_path(&self.timelineid, &self.tenantid));
+
+                // Fsync all the layer files and directory using multiple threads to
+                // minimize latency.
+                par_fsync::par_fsync(&layer_paths)?;
+
+                layer_paths.pop().unwrap();
             }
-
-            writer.as_mut().unwrap().put_value(key, lsn, value)?;
-            prev_key = Some(key);
         }
-        if let Some(writer) = writer {
-            new_layers.push(writer.finish(prev_key.unwrap().next())?);
-        }
-
-        // Sync layers
-        if !new_layers.is_empty() {
-            let mut layer_paths: Vec<PathBuf> = new_layers.iter().map(|l| l.path()).collect();
-
-            // also sync the directory
-            layer_paths.push(self.conf.timeline_path(&self.timelineid, &self.tenantid));
-
-            // Fsync all the layer files and directory using multiple threads to
-            // minimize latency.
-            par_fsync::par_fsync(&layer_paths)?;
-
-            layer_paths.pop().unwrap();
-        }
-
         let mut layers = self.layers.write().unwrap();
         for l in new_layers {
             layers.insert_historic(Arc::new(l));

pageserver/src/layered_repository/block_io.rs

@@ -73,7 +73,7 @@ pub struct BlockCursor<R>
 where
     R: BlockReader,
 {
-    reader: R,
+    pub reader: R,
     /// last accessed page
     cache: Option<(u32, R::BlockLease)>,
 }

pageserver/src/layered_repository/delta_layer.rs

@@ -23,6 +23,7 @@
 //! "values" part.  The actual page images and WAL records are stored in the
 //! "values" part.
 //!
+use crate::config;
 use crate::config::PageServerConf;
 use crate::layered_repository::blob_io::{BlobCursor, BlobWriter, WriteBlobWriter};
 use crate::layered_repository::block_io::{BlockBuf, BlockCursor, BlockReader, FileBlockReader};
@@ -39,7 +40,7 @@ use crate::{DELTA_FILE_MAGIC, STORAGE_FORMAT_VERSION};
 use anyhow::{bail, ensure, Context, Result};
 use serde::{Deserialize, Serialize};
 use tracing::*;
-// avoid binding to Write (conflicts with std::io::Write)
+// avoid binding to Write (conflicts with std::sio::Write)
 // while being able to use std::fmt::Write's methods
 use std::fmt::Write as _;
 use std::fs;
@@ -56,6 +57,8 @@ use utils::{
     zid::{ZTenantId, ZTimelineId},
 };
 
+const DICTIONARY_OFFSET: u64 = PAGE_SZ as u64;
+
 ///
 /// Header stored in the beginning of the file
 ///
@@ -195,6 +198,10 @@ pub struct DeltaLayerInner {
 
     /// Reader object for reading blocks from the file. (None if not loaded yet)
     file: Option<FileBlockReader<VirtualFile>>,
+
+    /// Compression dictionary.
+    dictionary: Vec<u8>, // empty if not loaded
+    prepared_dictionary: Option<zstd::dict::DecoderDictionary<'static>>, // None if not loaded
 }
 
 impl Layer for DeltaLayer {
@@ -232,6 +239,12 @@ impl Layer for DeltaLayer {
         {
             // Open the file and lock the metadata in memory
             let inner = self.load()?;
+            let mut decompressor = match &inner.prepared_dictionary {
+                Some(dictionary) => Some(zstd::bulk::Decompressor::with_prepared_dictionary(
+                    dictionary,
+                )?),
+                None => None,
+            };
 
             // Scan the page versions backwards, starting from `lsn`.
             let file = inner.file.as_ref().unwrap();
@@ -264,7 +277,14 @@ impl Layer for DeltaLayer {
                         file.file.path.display()
                     )
                 })?;
-                let val = Value::des(&buf).with_context(|| {
+                let val = if let Some(decompressor) = &mut decompressor {
+                    let decompressed =
+                        decompressor.decompress(&buf, config::ZSTD_DECOMPRESS_BUFFER_LIMIT)?;
+                    Value::des(&decompressed)
+                } else {
+                    Value::des(&buf)
+                }
+                .with_context(|| {
                     format!(
                         "Failed to deserialize file blob from virtual file {}",
                         file.file.path.display()
@@ -342,7 +362,6 @@ impl Layer for DeltaLayer {
         }
 
         let inner = self.load()?;
-
         println!(
             "index_start_blk: {}, root {}",
             inner.index_start_blk, inner.index_root_blk
@@ -358,6 +377,13 @@ impl Layer for DeltaLayer {
         tree_reader.dump()?;
 
         let mut cursor = file.block_cursor();
+        let mut decompressor = match &inner.prepared_dictionary {
+            Some(dictionary) => Some(zstd::bulk::Decompressor::with_prepared_dictionary(
+                dictionary,
+            )?),
+            None => None,
+        };
+
         tree_reader.visit(
             &[0u8; DELTA_KEY_SIZE],
             VisitDirection::Forwards,
@@ -369,7 +395,14 @@ impl Layer for DeltaLayer {
                 let mut desc = String::new();
                 match cursor.read_blob(blob_ref.pos()) {
                     Ok(buf) => {
-                        let val = Value::des(&buf);
+                        let val = if let Some(decompressor) = &mut decompressor {
+                            let decompressed = decompressor
+                                .decompress(&buf, config::ZSTD_DECOMPRESS_BUFFER_LIMIT)
+                                .unwrap();
+                            Value::des(&decompressed)
+                        } else {
+                            Value::des(&buf)
+                        };
                         match val {
                             Ok(Value::Image(img)) => {
                                 write!(&mut desc, " img {} bytes", img.len()).unwrap();
@@ -487,6 +520,13 @@ impl DeltaLayer {
             }
         }
 
+        let mut cursor = file.block_cursor();
+        inner.dictionary = cursor.read_blob(DICTIONARY_OFFSET)?;
+        inner.prepared_dictionary = if inner.dictionary.is_empty() {
+            None
+        } else {
+            Some(zstd::dict::DecoderDictionary::copy(&inner.dictionary))
+        };
         inner.index_start_blk = actual_summary.index_start_blk;
         inner.index_root_blk = actual_summary.index_root_blk;
 
@@ -512,6 +552,8 @@ impl DeltaLayer {
             inner: RwLock::new(DeltaLayerInner {
                 loaded: false,
                 file: None,
+                dictionary: Vec::new(),
+                prepared_dictionary: None,
                 index_start_blk: 0,
                 index_root_blk: 0,
             }),
@@ -539,6 +581,8 @@ impl DeltaLayer {
             inner: RwLock::new(DeltaLayerInner {
                 loaded: false,
                 file: None,
+                dictionary: Vec::new(),
+                prepared_dictionary: None,
                 index_start_blk: 0,
                 index_root_blk: 0,
             }),
@@ -586,6 +630,7 @@ pub struct DeltaLayerWriter {
     tree: DiskBtreeBuilder<BlockBuf, DELTA_KEY_SIZE>,
 
     blob_writer: WriteBlobWriter<BufWriter<VirtualFile>>,
+    compressor: Option<zstd::bulk::Compressor<'static>>,
 }
 
 impl DeltaLayerWriter {
@@ -598,6 +643,7 @@ impl DeltaLayerWriter {
         tenantid: ZTenantId,
         key_start: Key,
         lsn_range: Range<Lsn>,
+        dictionary: Vec<u8>,
     ) -> Result<DeltaLayerWriter> {
         // Create the file initially with a temporary filename. We don't know
         // the end key yet, so we cannot form the final filename yet. We will
@@ -613,14 +659,24 @@ impl DeltaLayerWriter {
         ));
         let mut file = VirtualFile::create(&path)?;
         // make room for the header block
-        file.seek(SeekFrom::Start(PAGE_SZ as u64))?;
+        file.seek(SeekFrom::Start(DICTIONARY_OFFSET))?;
         let buf_writer = BufWriter::new(file);
-        let blob_writer = WriteBlobWriter::new(buf_writer, PAGE_SZ as u64);
+        let mut blob_writer = WriteBlobWriter::new(buf_writer, DICTIONARY_OFFSET);
+
+        let off = blob_writer.write_blob(&dictionary)?;
+        assert!(off == DICTIONARY_OFFSET);
+        let compressor = if dictionary.is_empty() {
+            None
+        } else {
+            Some(zstd::bulk::Compressor::with_dictionary(
+                config::ZSTD_COMPRESSION_LEVEL,
+                &dictionary,
+            )?)
+        };
 
         // Initialize the b-tree index builder
         let block_buf = BlockBuf::new();
         let tree_builder = DiskBtreeBuilder::new(block_buf);
-
         Ok(DeltaLayerWriter {
             conf,
             path,
@@ -630,6 +686,7 @@ impl DeltaLayerWriter {
             lsn_range,
             tree: tree_builder,
             blob_writer,
+            compressor,
         })
     }
 
@@ -641,7 +698,13 @@ impl DeltaLayerWriter {
     pub fn put_value(&mut self, key: Key, lsn: Lsn, val: Value) -> Result<()> {
         assert!(self.lsn_range.start <= lsn);
 
-        let off = self.blob_writer.write_blob(&Value::ser(&val)?)?;
+        let body = &Value::ser(&val)?;
+        let off = if let Some(ref mut compressor) = self.compressor {
+            let compressed = compressor.compress(body)?;
+            self.blob_writer.write_blob(&compressed)?
+        } else {
+            self.blob_writer.write_blob(body)?
+        };
 
         let blob_ref = BlobRef::new(off, val.will_init());
 
@@ -698,6 +761,8 @@ impl DeltaLayerWriter {
             inner: RwLock::new(DeltaLayerInner {
                 loaded: false,
                 file: None,
+                dictionary: Vec::new(),
+                prepared_dictionary: None,
                 index_start_blk,
                 index_root_blk,
             }),
@@ -735,6 +800,7 @@ struct DeltaValueIter<'a> {
     all_offsets: Vec<(DeltaKey, BlobRef)>,
     next_idx: usize,
     reader: BlockCursor<Adapter<'a>>,
+    decompressor: Option<zstd::bulk::Decompressor<'a>>,
 }
 
 struct Adapter<'a>(RwLockReadGuard<'a, DeltaLayerInner>);
@@ -773,11 +839,26 @@ impl<'a> DeltaValueIter<'a> {
                 true
             },
         )?;
-
+        let decompressor = match &inner.prepared_dictionary {
+            Some(dictionary) => Some(zstd::bulk::Decompressor::with_prepared_dictionary(
+                dictionary,
+            )?),
+            None => None,
+        };
+        /*
+        let decompressor = if !inner.dictionary.is_empty() {
+            Some(zstd::bulk::Decompressor::with_dictionary(
+                &inner.dictionary,
+            )?)
+        } else {
+            None
+        };
+        */
         let iter = DeltaValueIter {
             all_offsets,
             next_idx: 0,
             reader: BlockCursor::new(Adapter(inner)),
+            decompressor,
         };
 
         Ok(iter)
@@ -791,7 +872,13 @@ impl<'a> DeltaValueIter<'a> {
             let lsn = delta_key.lsn();
 
             let buf = self.reader.read_blob(blob_ref.pos())?;
-            let val = Value::des(&buf)?;
+            let val = if let Some(decompressor) = &mut self.decompressor {
+                let decompressed =
+                    decompressor.decompress(&buf, config::ZSTD_DECOMPRESS_BUFFER_LIMIT)?;
+                Value::des(&decompressed)
+            } else {
+                Value::des(&buf)
+            }?;
             self.next_idx += 1;
             Ok(Some((key, lsn, val)))
         } else {

pageserver/src/layered_repository/disk_btree.rs

@@ -11,6 +11,7 @@
 //! - page-oriented
 //!
 //! TODO:
+//! - better errors (e.g. with thiserror?)
 //! - maybe something like an Adaptive Radix Tree would be more efficient?
 //! - the values stored by image and delta layers are offsets into the file,
 //!   and they are in monotonically increasing order. Prefix compression would
@@ -18,12 +19,11 @@
 //! - An Iterator interface would be more convenient for the callers than the
 //!   'visit' function
 //!
+use anyhow;
 use byteorder::{ReadBytesExt, BE};
 use bytes::{BufMut, Bytes, BytesMut};
 use hex;
-use std::{cmp::Ordering, io, result};
-use thiserror::Error;
-use tracing::error;
+use std::cmp::Ordering;
 
 use crate::layered_repository::block_io::{BlockReader, BlockWriter};
 
@@ -86,23 +86,6 @@ impl Value {
     }
 }
 
-#[derive(Error, Debug)]
-pub enum DiskBtreeError {
-    #[error("Attempt to append a value that is too large {0} > {}", MAX_VALUE)]
-    AppendOverflow(u64),
-
-    #[error("Unsorted input: key {key:?} is <= last_key {last_key:?}")]
-    UnsortedInput { key: Box<[u8]>, last_key: Box<[u8]> },
-
-    #[error("Could not push to new leaf node")]
-    FailedToPushToNewLeafNode,
-
-    #[error("IoError: {0}")]
-    Io(#[from] io::Error),
-}
-
-pub type Result<T> = result::Result<T, DiskBtreeError>;
-
 /// This is the on-disk representation.
 struct OnDiskNode<'a, const L: usize> {
     // Fixed-width fields
@@ -123,12 +106,12 @@ impl<'a, const L: usize> OnDiskNode<'a, L> {
     ///
     /// Interpret a PAGE_SZ page as a node.
     ///
-    fn deparse(buf: &[u8]) -> Result<OnDiskNode<L>> {
+    fn deparse(buf: &[u8]) -> OnDiskNode<L> {
         let mut cursor = std::io::Cursor::new(buf);
-        let num_children = cursor.read_u16::<BE>()?;
-        let level = cursor.read_u8()?;
-        let prefix_len = cursor.read_u8()?;
-        let suffix_len = cursor.read_u8()?;
+        let num_children = cursor.read_u16::<BE>().unwrap();
+        let level = cursor.read_u8().unwrap();
+        let prefix_len = cursor.read_u8().unwrap();
+        let suffix_len = cursor.read_u8().unwrap();
 
         let mut off = cursor.position();
         let prefix_off = off as usize;
@@ -146,7 +129,7 @@ impl<'a, const L: usize> OnDiskNode<'a, L> {
         let keys = &buf[keys_off..keys_off + keys_len];
         let values = &buf[values_off..values_off + values_len];
 
-        Ok(OnDiskNode {
+        OnDiskNode {
             num_children,
             level,
             prefix_len,
@@ -154,7 +137,7 @@ impl<'a, const L: usize> OnDiskNode<'a, L> {
             prefix,
             keys,
             values,
-        })
+        }
     }
 
     ///
@@ -166,11 +149,7 @@ impl<'a, const L: usize> OnDiskNode<'a, L> {
         Value::from_slice(value_slice)
     }
 
-    fn binary_search(
-        &self,
-        search_key: &[u8; L],
-        keybuf: &mut [u8],
-    ) -> result::Result<usize, usize> {
+    fn binary_search(&self, search_key: &[u8; L], keybuf: &mut [u8]) -> Result<usize, usize> {
         let mut size = self.num_children as usize;
         let mut low = 0;
         let mut high = size;
@@ -230,7 +209,7 @@ where
     ///
     /// Read the value for given key. Returns the value, or None if it doesn't exist.
     ///
-    pub fn get(&self, search_key: &[u8; L]) -> Result<Option<u64>> {
+    pub fn get(&self, search_key: &[u8; L]) -> anyhow::Result<Option<u64>> {
         let mut result: Option<u64> = None;
         self.visit(search_key, VisitDirection::Forwards, |key, value| {
             if key == search_key {
@@ -251,7 +230,7 @@ where
         search_key: &[u8; L],
         dir: VisitDirection,
         mut visitor: V,
-    ) -> Result<bool>
+    ) -> anyhow::Result<bool>
     where
         V: FnMut(&[u8], u64) -> bool,
     {
@@ -264,7 +243,7 @@ where
         search_key: &[u8; L],
         dir: VisitDirection,
         visitor: &mut V,
-    ) -> Result<bool>
+    ) -> anyhow::Result<bool>
     where
         V: FnMut(&[u8], u64) -> bool,
     {
@@ -281,11 +260,11 @@ where
         search_key: &[u8; L],
         dir: VisitDirection,
         visitor: &mut V,
-    ) -> Result<bool>
+    ) -> anyhow::Result<bool>
     where
         V: FnMut(&[u8], u64) -> bool,
     {
-        let node = OnDiskNode::deparse(node_buf)?;
+        let node = OnDiskNode::deparse(node_buf);
         let prefix_len = node.prefix_len as usize;
         let suffix_len = node.suffix_len as usize;
 
@@ -390,15 +369,15 @@ where
     }
 
     #[allow(dead_code)]
-    pub fn dump(&self) -> Result<()> {
+    pub fn dump(&self) -> anyhow::Result<()> {
         self.dump_recurse(self.root_blk, &[], 0)
     }
 
-    fn dump_recurse(&self, blknum: u32, path: &[u8], depth: usize) -> Result<()> {
+    fn dump_recurse(&self, blknum: u32, path: &[u8], depth: usize) -> anyhow::Result<()> {
         let blk = self.reader.read_blk(self.start_blk + blknum)?;
         let buf: &[u8] = blk.as_ref();
 
-        let node = OnDiskNode::<L>::deparse(buf)?;
+        let node = OnDiskNode::<L>::deparse(buf);
 
         print!("{:indent$}", "", indent = depth * 2);
         println!(
@@ -463,24 +442,17 @@ where
         }
     }
 
-    pub fn append(&mut self, key: &[u8; L], value: u64) -> Result<()> {
-        if value > MAX_VALUE {
-            return Err(DiskBtreeError::AppendOverflow(value));
-        }
+    pub fn append(&mut self, key: &[u8; L], value: u64) -> Result<(), anyhow::Error> {
+        assert!(value <= MAX_VALUE);
         if let Some(last_key) = &self.last_key {
-            if key <= last_key {
-                return Err(DiskBtreeError::UnsortedInput {
-                    key: key.as_slice().into(),
-                    last_key: last_key.as_slice().into(),
-                });
-            }
+            assert!(key > last_key, "unsorted input");
         }
         self.last_key = Some(*key);
 
-        self.append_internal(key, Value::from_u64(value))
+        Ok(self.append_internal(key, Value::from_u64(value))?)
     }
 
-    fn append_internal(&mut self, key: &[u8; L], value: Value) -> Result<()> {
+    fn append_internal(&mut self, key: &[u8; L], value: Value) -> Result<(), std::io::Error> {
         // Try to append to the current leaf buffer
         let last = self.stack.last_mut().unwrap();
         let level = last.level;
@@ -504,15 +476,14 @@ where
         // key to it.
         let mut last = BuildNode::new(level);
         if !last.push(key, value) {
-            return Err(DiskBtreeError::FailedToPushToNewLeafNode);
+            panic!("could not push to new leaf node");
         }
-
         self.stack.push(last);
 
         Ok(())
     }
 
-    fn flush_node(&mut self) -> Result<()> {
+    fn flush_node(&mut self) -> Result<(), std::io::Error> {
         let last = self.stack.pop().unwrap();
         let buf = last.pack();
         let downlink_key = last.first_key();
@@ -534,7 +505,7 @@ where
     /// (In the image and delta layers, it is stored in the beginning of the file,
     /// in the summary header)
     ///
-    pub fn finish(mut self) -> Result<(u32, W)> {
+    pub fn finish(mut self) -> Result<(u32, W), std::io::Error> {
         // flush all levels, except the root.
         while self.stack.len() > 1 {
             self.flush_node()?;
@@ -721,14 +692,14 @@ mod tests {
     impl BlockReader for TestDisk {
         type BlockLease = std::rc::Rc<[u8; PAGE_SZ]>;
 
-        fn read_blk(&self, blknum: u32) -> io::Result<Self::BlockLease> {
+        fn read_blk(&self, blknum: u32) -> Result<Self::BlockLease, std::io::Error> {
             let mut buf = [0u8; PAGE_SZ];
             buf.copy_from_slice(&self.blocks[blknum as usize]);
             Ok(std::rc::Rc::new(buf))
         }
     }
     impl BlockWriter for &mut TestDisk {
-        fn write_blk(&mut self, buf: Bytes) -> io::Result<u32> {
+        fn write_blk(&mut self, buf: Bytes) -> Result<u32, std::io::Error> {
             let blknum = self.blocks.len();
             self.blocks.push(buf);
             Ok(blknum as u32)
@@ -736,7 +707,7 @@ mod tests {
     }
 
     #[test]
-    fn basic() -> Result<()> {
+    fn basic() -> anyhow::Result<()> {
         let mut disk = TestDisk::new();
         let mut writer = DiskBtreeBuilder::<_, 6>::new(&mut disk);
 
@@ -817,7 +788,7 @@ mod tests {
     }
 
     #[test]
-    fn lots_of_keys() -> Result<()> {
+    fn lots_of_keys() -> anyhow::Result<()> {
         let mut disk = TestDisk::new();
         let mut writer = DiskBtreeBuilder::<_, 8>::new(&mut disk);
 
@@ -911,7 +882,7 @@ mod tests {
     }
 
     #[test]
-    fn random_data() -> Result<()> {
+    fn random_data() -> anyhow::Result<()> {
         // Generate random keys with exponential distribution, to
         // exercise the prefix compression
         const NUM_KEYS: usize = 100000;
@@ -956,27 +927,21 @@ mod tests {
     }
 
     #[test]
+    #[should_panic(expected = "unsorted input")]
     fn unsorted_input() {
         let mut disk = TestDisk::new();
         let mut writer = DiskBtreeBuilder::<_, 2>::new(&mut disk);
 
         let _ = writer.append(b"ba", 1);
         let _ = writer.append(b"bb", 2);
-        let err = writer.append(b"aa", 3).expect_err("should've failed");
-        match err {
-            DiskBtreeError::UnsortedInput { key, last_key } => {
-                assert_eq!(key.as_ref(), b"aa".as_slice());
-                assert_eq!(last_key.as_ref(), b"bb".as_slice());
-            }
-            _ => panic!("unexpected error variant, expected DiskBtreeError::UnsortedInput"),
-        }
+        let _ = writer.append(b"aa", 3);
     }
 
     ///
     /// This test contains a particular data set, see disk_btree_test_data.rs
     ///
     #[test]
-    fn particular_data() -> Result<()> {
+    fn particular_data() -> anyhow::Result<()> {
         // Build a tree from it
         let mut disk = TestDisk::new();
         let mut writer = DiskBtreeBuilder::<_, 26>::new(&mut disk);

pageserver/src/layered_repository/image_layer.rs

@@ -19,6 +19,7 @@
 //! layer, and offsets to the other parts. The "index" is a B-tree,
 //! mapping from Key to an offset in the "values" part.  The
 //! actual page images are stored in the "values" part.
+use crate::config;
 use crate::config::PageServerConf;
 use crate::layered_repository::blob_io::{BlobCursor, BlobWriter, WriteBlobWriter};
 use crate::layered_repository::block_io::{BlockBuf, BlockReader, FileBlockReader};
@@ -168,7 +169,8 @@ impl Layer for ImageLayer {
                     offset
                 )
             })?;
-            let value = Bytes::from(blob);
+            let decompressed = zstd::bulk::decompress(&blob, PAGE_SZ)?;
+            let value = Bytes::from(decompressed);
 
             reconstruct_state.img = Some((self.lsn, value));
             Ok(ValueReconstructResult::Complete)
@@ -456,7 +458,8 @@ impl ImageLayerWriter {
     ///
     pub fn put_image(&mut self, key: Key, img: &[u8]) -> Result<()> {
         ensure!(self.key_range.contains(&key));
-        let off = self.blob_writer.write_blob(img)?;
+        let compressed = zstd::bulk::compress(img, config::ZSTD_COMPRESSION_LEVEL)?;
+        let off = self.blob_writer.write_blob(&compressed)?;
 
         let mut keybuf: [u8; KEY_SIZE] = [0u8; KEY_SIZE];
         key.write_to_byte_slice(&mut keybuf);

pageserver/src/layered_repository/inmemory_layer.rs

@@ -4,6 +4,7 @@
 //! held in an ephemeral file, not in memory. The metadata for each page version, i.e.
 //! its position in the file, is kept in memory, though.
 //!
+use crate::config;
 use crate::config::PageServerConf;
 use crate::layered_repository::blob_io::{BlobCursor, BlobWriter};
 use crate::layered_repository::block_io::BlockReader;
@@ -318,21 +319,40 @@ impl InMemoryLayer {
         // rare though, so we just accept the potential latency hit for now.
         let inner = self.inner.read().unwrap();
 
+        let mut samples: Vec<Vec<u8>> = Vec::with_capacity(config::ZSTD_MAX_SAMPLES);
+        let mut buf = Vec::new();
+
+        let mut keys: Vec<(&Key, &VecMap<Lsn, u64>)> = inner.index.iter().collect();
+        keys.sort_by_key(|k| k.0);
+
+        let mut cursor = inner.file.block_cursor();
+
+        // First learn dictionary
+        'train: for (_key, vec_map) in keys.iter() {
+            // Write all page versions
+            for (_lsn, pos) in vec_map.as_slice() {
+                cursor.read_blob_into_buf(*pos, &mut buf)?;
+                samples.push(buf.clone());
+                if samples.len() == config::ZSTD_MAX_SAMPLES {
+                    break 'train;
+                }
+            }
+        }
+
+        let dictionary = if samples.len() >= config::ZSTD_MIN_SAMPLES {
+            zstd::dict::from_samples(&samples, config::ZSTD_MAX_DICTIONARY_SIZE)?
+        } else {
+            Vec::new()
+        };
         let mut delta_layer_writer = DeltaLayerWriter::new(
             self.conf,
             self.timelineid,
             self.tenantid,
             Key::MIN,
             self.start_lsn..inner.end_lsn.unwrap(),
+            dictionary,
         )?;
 
-        let mut buf = Vec::new();
-
-        let mut cursor = inner.file.block_cursor();
-
-        let mut keys: Vec<(&Key, &VecMap<Lsn, u64>)> = inner.index.iter().collect();
-        keys.sort_by_key(|k| k.0);
-
         for (key, vec_map) in keys.iter() {
             let key = **key;
             // Write all page versions

pageserver/src/layered_repository/layer_map.rs

@@ -43,13 +43,10 @@ pub struct LayerMap {
     pub next_open_layer_at: Option<Lsn>,
 
     ///
-    /// Frozen layers, if any. Frozen layers are in-memory layers that
-    /// are no longer added to, but haven't been written out to disk
-    /// yet. They contain WAL older than the current 'open_layer' or
-    /// 'next_open_layer_at', but newer than any historic layer.
-    /// The frozen layers are in order from oldest to newest, so that
-    /// the newest one is in the 'back' of the VecDeque, and the oldest
-    /// in the 'front'.
+    /// The frozen layer, if any, contains WAL older than the current 'open_layer'
+    /// or 'next_open_layer_at', but newer than any historic layer. The frozen
+    /// layer is during checkpointing, when an InMemoryLayer is being written out
+    /// to disk.
     ///
     pub frozen_layers: VecDeque<Arc<InMemoryLayer>>,
 

pageserver/src/lib.rs

@@ -67,7 +67,7 @@ pub type RepositoryImpl = LayeredRepository;
 
 pub type DatadirTimelineImpl = DatadirTimeline<RepositoryImpl>;
 
-pub fn shutdown_pageserver(exit_code: i32) {
+pub fn shutdown_pageserver() {
     // Shut down the libpq endpoint thread. This prevents new connections from
     // being accepted.
     thread_mgr::shutdown_threads(Some(ThreadKind::LibpqEndpointListener), None, None);
@@ -94,5 +94,5 @@ pub fn shutdown_pageserver(exit_code: i32) {
     thread_mgr::shutdown_threads(None, None, None);
 
     info!("Shut down successfully completed");
-    std::process::exit(exit_code);
+    std::process::exit(0);
 }

pageserver/src/page_service.rs

@@ -31,7 +31,7 @@ use utils::{
 
 use crate::basebackup;
 use crate::config::{PageServerConf, ProfilingConfig};
-use crate::pgdatadir_mapping::{DatadirTimeline, LsnForTimestamp};
+use crate::pgdatadir_mapping::DatadirTimeline;
 use crate::profiling::profpoint_start;
 use crate::reltag::RelTag;
 use crate::repository::Repository;
@@ -42,7 +42,6 @@ use crate::thread_mgr::ThreadKind;
 use crate::walreceiver;
 use crate::CheckpointConfig;
 use metrics::{register_histogram_vec, HistogramVec};
-use postgres_ffi::xlog_utils::to_pg_timestamp;
 
 // Wrapped in libpq CopyData
 enum PagestreamFeMessage {
@@ -806,33 +805,6 @@ impl postgres_backend::Handler for PageServerHandler {
 
             pgb.write_message_noflush(&SINGLE_COL_ROWDESC)?
                 .write_message_noflush(&BeMessage::CommandComplete(b"SELECT 1"))?;
-        } else if query_string.starts_with("get_lsn_by_timestamp ") {
-            // Locate LSN of last transaction with timestamp less or equal than sppecified
-            // TODO lazy static
-            let re = Regex::new(r"^get_lsn_by_timestamp ([[:xdigit:]]+) ([[:xdigit:]]+) '(.*)'$")
-                .unwrap();
-            let caps = re
-                .captures(query_string)
-                .with_context(|| format!("invalid get_lsn_by_timestamp: '{}'", query_string))?;
-
-            let tenantid = ZTenantId::from_str(caps.get(1).unwrap().as_str())?;
-            let timelineid = ZTimelineId::from_str(caps.get(2).unwrap().as_str())?;
-            let timeline = tenant_mgr::get_local_timeline_with_load(tenantid, timelineid)
-                .context("Cannot load local timeline")?;
-
-            let timestamp = humantime::parse_rfc3339(caps.get(3).unwrap().as_str())?;
-            let timestamp_pg = to_pg_timestamp(timestamp);
-
-            pgb.write_message_noflush(&BeMessage::RowDescription(&[RowDescriptor::text_col(
-                b"lsn",
-            )]))?;
-            let result = match timeline.find_lsn_for_timestamp(timestamp_pg)? {
-                LsnForTimestamp::Present(lsn) => format!("{}", lsn),
-                LsnForTimestamp::Future(_lsn) => "future".into(),
-                LsnForTimestamp::Past(_lsn) => "past".into(),
-            };
-            pgb.write_message_noflush(&BeMessage::DataRow(&[Some(result.as_bytes())]))?;
-            pgb.write_message(&BeMessage::CommandComplete(b"SELECT 1"))?;
         } else {
             bail!("unknown command");
         }

pageserver/src/pgdatadir_mapping.rs

@@ -13,7 +13,6 @@ use crate::repository::{Repository, Timeline};
 use crate::walrecord::ZenithWalRecord;
 use anyhow::{bail, ensure, Result};
 use bytes::{Buf, Bytes};
-use postgres_ffi::xlog_utils::TimestampTz;
 use postgres_ffi::{pg_constants, Oid, TransactionId};
 use serde::{Deserialize, Serialize};
 use std::collections::{HashMap, HashSet};
@@ -46,13 +45,6 @@ where
     current_logical_size: AtomicIsize,
 }
 
-#[derive(Debug)]
-pub enum LsnForTimestamp {
-    Present(Lsn),
-    Future(Lsn),
-    Past(Lsn),
-}
-
 impl<R: Repository> DatadirTimeline<R> {
     pub fn new(tline: Arc<R::Timeline>, repartition_threshold: u64) -> Self {
         DatadirTimeline {
@@ -210,106 +202,6 @@ impl<R: Repository> DatadirTimeline<R> {
         Ok(exists)
     }
 
-    /// Locate LSN, such that all transactions that committed before
-    /// 'search_timestamp' are visible, but nothing newer is.
-    ///
-    /// This is not exact. Commit timestamps are not guaranteed to be ordered,
-    /// so it's not well defined which LSN you get if there were multiple commits
-    /// "in flight" at that point in time.
-    ///
-    pub fn find_lsn_for_timestamp(&self, search_timestamp: TimestampTz) -> Result<LsnForTimestamp> {
-        let gc_cutoff_lsn_guard = self.tline.get_latest_gc_cutoff_lsn();
-        let min_lsn = *gc_cutoff_lsn_guard;
-        let max_lsn = self.tline.get_last_record_lsn();
-
-        // LSNs are always 8-byte aligned. low/mid/high represent the
-        // LSN divided by 8.
-        let mut low = min_lsn.0 / 8;
-        let mut high = max_lsn.0 / 8 + 1;
-
-        let mut found_smaller = false;
-        let mut found_larger = false;
-        while low < high {
-            // cannot overflow, high and low are both smaller than u64::MAX / 2
-            let mid = (high + low) / 2;
-
-            let cmp = self.is_latest_commit_timestamp_ge_than(
-                search_timestamp,
-                Lsn(mid * 8),
-                &mut found_smaller,
-                &mut found_larger,
-            )?;
-
-            if cmp {
-                high = mid;
-            } else {
-                low = mid + 1;
-            }
-        }
-        match (found_smaller, found_larger) {
-            (false, false) => {
-                // This can happen if no commit records have been processed yet, e.g.
-                // just after importing a cluster.
-                bail!("no commit timestamps found");
-            }
-            (true, false) => {
-                // Didn't find any commit timestamps larger than the request
-                Ok(LsnForTimestamp::Future(max_lsn))
-            }
-            (false, true) => {
-                // Didn't find any commit timestamps smaller than the request
-                Ok(LsnForTimestamp::Past(max_lsn))
-            }
-            (true, true) => {
-                // low is the LSN of the first commit record *after* the search_timestamp,
-                // Back off by one to get to the point just before the commit.
-                //
-                // FIXME: it would be better to get the LSN of the previous commit.
-                // Otherwise, if you restore to the returned LSN, the database will
-                // include physical changes from later commits that will be marked
-                // as aborted, and will need to be vacuumed away.
-                Ok(LsnForTimestamp::Present(Lsn((low - 1) * 8)))
-            }
-        }
-    }
-
-    ///
-    /// Subroutine of find_lsn_for_timestamp(). Returns true, if there are any
-    /// commits that committed after 'search_timestamp', at LSN 'probe_lsn'.
-    ///
-    /// Additionally, sets 'found_smaller'/'found_Larger, if encounters any commits
-    /// with a smaller/larger timestamp.
-    ///
-    fn is_latest_commit_timestamp_ge_than(
-        &self,
-        search_timestamp: TimestampTz,
-        probe_lsn: Lsn,
-        found_smaller: &mut bool,
-        found_larger: &mut bool,
-    ) -> Result<bool> {
-        for segno in self.list_slru_segments(SlruKind::Clog, probe_lsn)? {
-            let nblocks = self.get_slru_segment_size(SlruKind::Clog, segno, probe_lsn)?;
-            for blknum in (0..nblocks).rev() {
-                let clog_page =
-                    self.get_slru_page_at_lsn(SlruKind::Clog, segno, blknum, probe_lsn)?;
-
-                if clog_page.len() == pg_constants::BLCKSZ as usize + 8 {
-                    let mut timestamp_bytes = [0u8; 8];
-                    timestamp_bytes.copy_from_slice(&clog_page[pg_constants::BLCKSZ as usize..]);
-                    let timestamp = TimestampTz::from_be_bytes(timestamp_bytes);
-
-                    if timestamp >= search_timestamp {
-                        *found_larger = true;
-                        return Ok(true);
-                    } else {
-                        *found_smaller = true;
-                    }
-                }
-            }
-        }
-        Ok(false)
-    }
-
     /// Get a list of SLRU segments
     pub fn list_slru_segments(&self, kind: SlruKind, lsn: Lsn) -> Result<HashSet<u32>> {
         // fetch directory entry

pageserver/src/thread_mgr.rs

@@ -130,14 +130,12 @@ struct PageServerThread {
 }
 
 /// Launch a new thread
-/// Note: if shutdown_process_on_error is set to true failure
-///   of the thread will lead to shutdown of entire process
 pub fn spawn<F>(
     kind: ThreadKind,
     tenant_id: Option<ZTenantId>,
     timeline_id: Option<ZTimelineId>,
     name: &str,
-    shutdown_process_on_error: bool,
+    fail_on_error: bool,
     f: F,
 ) -> std::io::Result<()>
 where
@@ -177,7 +175,7 @@ where
                 thread_id,
                 thread_rc2,
                 shutdown_rx,
-                shutdown_process_on_error,
+                fail_on_error,
                 f,
             )
         }) {
@@ -203,7 +201,7 @@ fn thread_wrapper<F>(
     thread_id: u64,
     thread: Arc<PageServerThread>,
     shutdown_rx: watch::Receiver<()>,
-    shutdown_process_on_error: bool,
+    fail_on_error: bool,
     f: F,
 ) where
     F: FnOnce() -> anyhow::Result<()> + Send + 'static,
@@ -223,41 +221,27 @@ fn thread_wrapper<F>(
     let result = panic::catch_unwind(AssertUnwindSafe(f));
 
     // Remove our entry from the global hashmap.
-    let thread = THREADS
-        .lock()
-        .unwrap()
-        .remove(&thread_id)
-        .expect("no thread in registry");
+    THREADS.lock().unwrap().remove(&thread_id);
 
     match result {
         Ok(Ok(())) => debug!("Thread '{}' exited normally", thread_name),
         Ok(Err(err)) => {
-            if shutdown_process_on_error {
+            if fail_on_error {
                 error!(
-                    "Shutting down: thread '{}' tenant_id: {:?}, timeline_id: {:?} exited with error: {:?}",
-                    thread_name, thread.tenant_id, thread.timeline_id, err
+                    "Shutting down: thread '{}' exited with error: {:?}",
+                    thread_name, err
                 );
-                shutdown_pageserver(1);
+                shutdown_pageserver();
             } else {
-                error!(
-                    "Thread '{}' tenant_id: {:?}, timeline_id: {:?} exited with error: {:?}",
-                    thread_name, thread.tenant_id, thread.timeline_id, err
-                );
+                error!("Thread '{}' exited with error: {:?}", thread_name, err);
             }
         }
         Err(err) => {
-            if shutdown_process_on_error {
-                error!(
-                    "Shutting down: thread '{}' tenant_id: {:?}, timeline_id: {:?} panicked: {:?}",
-                    thread_name, thread.tenant_id, thread.timeline_id, err
-                );
-                shutdown_pageserver(1);
-            } else {
-                error!(
-                    "Thread '{}' tenant_id: {:?}, timeline_id: {:?} panicked: {:?}",
-                    thread_name, thread.tenant_id, thread.timeline_id, err
-                );
-            }
+            error!(
+                "Shutting down: thread '{}' panicked: {:?}",
+                thread_name, err
+            );
+            shutdown_pageserver();
         }
     }
 }

pageserver/src/walingest.rs

@@ -635,10 +635,7 @@ impl<'a, R: Repository> WalIngest<'a, R> {
                     segno,
                     rpageno,
                     if is_commit {
-                        ZenithWalRecord::ClogSetCommitted {
-                            xids: page_xids,
-                            timestamp: parsed.xact_time,
-                        }
+                        ZenithWalRecord::ClogSetCommitted { xids: page_xids }
                     } else {
                         ZenithWalRecord::ClogSetAborted { xids: page_xids }
                     },
@@ -655,10 +652,7 @@ impl<'a, R: Repository> WalIngest<'a, R> {
             segno,
             rpageno,
             if is_commit {
-                ZenithWalRecord::ClogSetCommitted {
-                    xids: page_xids,
-                    timestamp: parsed.xact_time,
-                }
+                ZenithWalRecord::ClogSetCommitted { xids: page_xids }
             } else {
                 ZenithWalRecord::ClogSetAborted { xids: page_xids }
             },

pageserver/src/walrecord.rs

@@ -24,10 +24,7 @@ pub enum ZenithWalRecord {
         flags: u8,
     },
     /// Mark transaction IDs as committed on a CLOG page
-    ClogSetCommitted {
-        xids: Vec<TransactionId>,
-        timestamp: TimestampTz,
-    },
+    ClogSetCommitted { xids: Vec<TransactionId> },
     /// Mark transaction IDs as aborted on a CLOG page
     ClogSetAborted { xids: Vec<TransactionId> },
     /// Extend multixact offsets SLRU

pageserver/src/walredo.rs

@@ -283,11 +283,6 @@ impl PostgresRedoManager {
         // If something went wrong, don't try to reuse the process. Kill it, and
         // next request will launch a new one.
         if result.is_err() {
-            error!(
-                "error applying {} WAL records to reconstruct page image at LSN {}",
-                records.len(),
-                lsn
-            );
             let process = process_guard.take().unwrap();
             process.kill();
         }
@@ -392,7 +387,7 @@ impl PostgresRedoManager {
             }
             // Non-relational WAL records are handled here, with custom code that has the
             // same effects as the corresponding Postgres WAL redo function.
-            ZenithWalRecord::ClogSetCommitted { xids, timestamp } => {
+            ZenithWalRecord::ClogSetCommitted { xids } => {
                 let (slru_kind, segno, blknum) =
                     key_to_slru_block(key).or(Err(WalRedoError::InvalidRecord))?;
                 assert_eq!(
@@ -426,21 +421,6 @@ impl PostgresRedoManager {
                         page,
                     );
                 }
-
-                // Append the timestamp
-                if page.len() == pg_constants::BLCKSZ as usize + 8 {
-                    page.truncate(pg_constants::BLCKSZ as usize);
-                }
-                if page.len() == pg_constants::BLCKSZ as usize {
-                    page.extend_from_slice(&timestamp.to_be_bytes());
-                } else {
-                    warn!(
-                        "CLOG blk {} in seg {} has invalid size {}",
-                        blknum,
-                        segno,
-                        page.len()
-                    );
-                }
             }
             ZenithWalRecord::ClogSetAborted { xids } => {
                 let (slru_kind, segno, blknum) =

proxy/Cargo.toml

@@ -5,7 +5,6 @@ edition = "2021"
 
 [dependencies]
 anyhow = "1.0"
-async-trait = "0.1"
 base64 = "0.13.0"
 bytes = { version = "1.0.1", features = ['serde'] }
 clap = "3.0"
@@ -32,13 +31,13 @@ thiserror = "1.0.30"
 tokio = { version = "1.17", features = ["macros"] }
 tokio-postgres = { git = "https://github.com/zenithdb/rust-postgres.git", rev="d052ee8b86fff9897c77b0fe89ea9daba0e1fa38" }
 tokio-rustls = "0.23.0"
-url = "2.2.2"
 
 utils = { path = "../libs/utils" }
 metrics = { path = "../libs/metrics" }
 workspace_hack = { version = "0.1", path = "../workspace_hack" }
 
 [dev-dependencies]
+async-trait = "0.1"
 rcgen = "0.8.14"
 rstest = "0.12"
 tokio-postgres-rustls = "0.9.0"

proxy/README.md

@@ -1,33 +0,0 @@
-# Proxy
-
-Proxy binary accepts `--auth-backend` CLI option, which determines auth scheme and cluster routing method. Following backends are currently implemented:
-
-* legacy
-  old method, when username ends with `@zenith` it uses md5 auth dbname as the cluster name; otherwise, it sends a login link and waits for the console to call back
-* console
-  new SCRAM-based console API; uses SNI info to select the destination cluster
-* postgres
-  uses postgres to select auth secrets of existing roles. Useful for local testing
-* link
-  sends login link for all usernames
-
-## Using SNI-based routing on localhost
-
-Now proxy determines cluster name from the subdomain, request to the `my-cluster-42.somedomain.tld` will be routed to the cluster named `my-cluster-42`. Unfortunately `/etc/hosts` does not support domain wildcards, so I usually use `*.localtest.me` which resolves to `127.0.0.1`. Now we can create self-signed certificate and play with proxy:
-
-```
-openssl req -new -x509 -days 365 -nodes -text -out server.crt -keyout server.key -subj "/CN=*.localtest.me"
-
-```
-
-now you can start proxy:
-
-```
-./target/debug/proxy -c server.crt -k server.key
-```
-
-and connect to it:
-
-```
-PGSSLROOTCERT=./server.crt psql 'postgres://my-cluster-42.localtest.me:1234?sslmode=verify-full'
-```

proxy/src/auth.rs

@@ -1,17 +1,22 @@
 mod credentials;
+
+#[cfg(test)]
 mod flow;
 
-use crate::auth_backend::{console, legacy_console, link, postgres};
-use crate::config::{AuthBackendType, ProxyConfig};
+use crate::compute::DatabaseInfo;
+use crate::config::ProxyConfig;
+use crate::cplane_api::{self, CPlaneApi};
 use crate::error::UserFacingError;
 use crate::stream::PqStream;
-use crate::{auth_backend, compute, waiters};
-use console::ConsoleAuthError::SniMissing;
+use crate::waiters;
 use std::io;
 use thiserror::Error;
 use tokio::io::{AsyncRead, AsyncWrite};
+use utils::pq_proto::{BeMessage as Be, BeParameterStatusMessage};
 
 pub use credentials::ClientCredentials;
+
+#[cfg(test)]
 pub use flow::*;
 
 /// Common authentication error.
@@ -19,11 +24,9 @@ pub use flow::*;
 pub enum AuthErrorImpl {
     /// Authentication error reported by the console.
     #[error(transparent)]
-    Console(#[from] auth_backend::AuthError),
-
-    #[error(transparent)]
-    GetAuthInfo(#[from] auth_backend::console::ConsoleAuthError),
+    Console(#[from] cplane_api::AuthError),
 
+    #[cfg(test)]
     #[error(transparent)]
     Sasl(#[from] crate::sasl::Error),
 
@@ -38,19 +41,19 @@ pub enum AuthErrorImpl {
 
 impl AuthErrorImpl {
     pub fn auth_failed(msg: impl Into<String>) -> Self {
-        AuthErrorImpl::Console(auth_backend::AuthError::auth_failed(msg))
+        AuthErrorImpl::Console(cplane_api::AuthError::auth_failed(msg))
     }
 }
 
 impl From<waiters::RegisterError> for AuthErrorImpl {
     fn from(e: waiters::RegisterError) -> Self {
-        AuthErrorImpl::Console(auth_backend::AuthError::from(e))
+        AuthErrorImpl::Console(cplane_api::AuthError::from(e))
     }
 }
 
 impl From<waiters::WaitError> for AuthErrorImpl {
     fn from(e: waiters::WaitError) -> Self {
-        AuthErrorImpl::Console(auth_backend::AuthError::from(e))
+        AuthErrorImpl::Console(cplane_api::AuthError::from(e))
     }
 }
 
@@ -73,33 +76,112 @@ impl UserFacingError for AuthError {
         match self.0.as_ref() {
             Console(e) => e.to_string_client(),
             MalformedPassword => self.to_string(),
-            GetAuthInfo(e) if matches!(e, SniMissing) => e.to_string(),
             _ => "Internal error".to_string(),
         }
     }
 }
 
-async fn handle_user(
-    config: &ProxyConfig,
-    client: &mut PqStream<impl AsyncRead + AsyncWrite + Unpin + Send>,
+async fn handle_static(
+    host: String,
+    port: u16,
+    client: &mut PqStream<impl AsyncRead + AsyncWrite + Unpin>,
     creds: ClientCredentials,
-) -> Result<compute::NodeInfo, AuthError> {
-    match config.auth_backend {
-        AuthBackendType::LegacyConsole => {
-            legacy_console::handle_user(
-                &config.auth_endpoint,
-                &config.auth_link_uri,
-                client,
-                &creds,
-            )
-            .await
-        }
-        AuthBackendType::Console => {
-            console::handle_user(config.auth_endpoint.as_ref(), client, &creds).await
-        }
-        AuthBackendType::Postgres => {
-            postgres::handle_user(&config.auth_endpoint, client, &creds).await
-        }
-        AuthBackendType::Link => link::handle_user(config.auth_link_uri.as_ref(), client).await,
-    }
+) -> Result<DatabaseInfo, AuthError> {
+    client
+        .write_message(&Be::AuthenticationCleartextPassword)
+        .await?;
+
+    // Read client's password bytes
+    let msg = client.read_password_message().await?;
+    let cleartext_password = parse_password(&msg).ok_or(AuthErrorImpl::MalformedPassword)?;
+
+    let db_info = DatabaseInfo {
+        host,
+        port,
+        dbname: creds.dbname.clone(),
+        user: creds.user.clone(),
+        password: Some(cleartext_password.into()),
+    };
+
+    client
+        .write_message_noflush(&Be::AuthenticationOk)?
+        .write_message_noflush(&BeParameterStatusMessage::encoding())?;
+
+    Ok(db_info)
+}
+
+async fn handle_existing_user(
+    config: &ProxyConfig,
+    client: &mut PqStream<impl AsyncRead + AsyncWrite + Unpin>,
+    creds: ClientCredentials,
+) -> Result<DatabaseInfo, AuthError> {
+    let psql_session_id = new_psql_session_id();
+    let md5_salt = rand::random();
+
+    client
+        .write_message(&Be::AuthenticationMD5Password(md5_salt))
+        .await?;
+
+    // Read client's password hash
+    let msg = client.read_password_message().await?;
+    let md5_response = parse_password(&msg).ok_or(AuthErrorImpl::MalformedPassword)?;
+
+    let cplane = CPlaneApi::new(config.auth_endpoint.clone());
+    let db_info = cplane
+        .authenticate_proxy_client(creds, md5_response, &md5_salt, &psql_session_id)
+        .await?;
+
+    client
+        .write_message_noflush(&Be::AuthenticationOk)?
+        .write_message_noflush(&BeParameterStatusMessage::encoding())?;
+
+    Ok(db_info)
+}
+
+async fn handle_new_user(
+    config: &ProxyConfig,
+    client: &mut PqStream<impl AsyncRead + AsyncWrite + Unpin>,
+) -> Result<DatabaseInfo, AuthError> {
+    let psql_session_id = new_psql_session_id();
+    let greeting = hello_message(&config.redirect_uri, &psql_session_id);
+
+    let db_info = cplane_api::with_waiter(psql_session_id, |waiter| async {
+        // Give user a URL to spawn a new database
+        client
+            .write_message_noflush(&Be::AuthenticationOk)?
+            .write_message_noflush(&BeParameterStatusMessage::encoding())?
+            .write_message(&Be::NoticeResponse(&greeting))
+            .await?;
+
+        // Wait for web console response (see `mgmt`)
+        waiter.await?.map_err(AuthErrorImpl::auth_failed)
+    })
+    .await?;
+
+    client.write_message_noflush(&Be::NoticeResponse("Connecting to database."))?;
+
+    Ok(db_info)
+}
+
+fn new_psql_session_id() -> String {
+    hex::encode(rand::random::<[u8; 8]>())
+}
+
+fn parse_password(bytes: &[u8]) -> Option<&str> {
+    std::str::from_utf8(bytes).ok()?.strip_suffix('\0')
+}
+
+fn hello_message(redirect_uri: &str, session_id: &str) -> String {
+    format!(
+        concat![
+            "☀️  Welcome to Neon!\n",
+            "To proceed with database creation, open the following link:\n\n",
+            "    {redirect_uri}{session_id}\n\n",
+            "It needs to be done once and we will send you '.pgpass' file,\n",
+            "which will allow you to access or create ",
+            "databases without opening your web browser."
+        ],
+        redirect_uri = redirect_uri,
+        session_id = session_id,
+    )
 }

proxy/src/auth/credentials.rs

@@ -1,7 +1,7 @@
 //! User credentials used in authentication.
 
 use super::AuthError;
-use crate::compute;
+use crate::compute::DatabaseInfo;
 use crate::config::ProxyConfig;
 use crate::error::UserFacingError;
 use crate::stream::PqStream;
@@ -18,22 +18,10 @@ pub enum ClientCredsParseError {
 impl UserFacingError for ClientCredsParseError {}
 
 /// Various client credentials which we use for authentication.
-/// Note that we don't store any kind of client key or password here.
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, PartialEq, Eq)]
 pub struct ClientCredentials {
     pub user: String,
     pub dbname: String,
-
-    // New console API requires SNI info to determine the cluster name.
-    // Other Auth backends don't need it.
-    pub sni_data: Option<String>,
-}
-
-impl ClientCredentials {
-    pub fn is_existing_user(&self) -> bool {
-        // This logic will likely change in the future.
-        self.user.ends_with("@zenith")
-    }
 }
 
 impl TryFrom<HashMap<String, String>> for ClientCredentials {
@@ -49,11 +37,7 @@ impl TryFrom<HashMap<String, String>> for ClientCredentials {
         let user = get_param("user")?;
         let db = get_param("database")?;
 
-        Ok(Self {
-            user,
-            dbname: db,
-            sni_data: None,
-        })
+        Ok(Self { user, dbname: db })
     }
 }
 
@@ -62,9 +46,21 @@ impl ClientCredentials {
     pub async fn authenticate(
         self,
         config: &ProxyConfig,
-        client: &mut PqStream<impl AsyncRead + AsyncWrite + Unpin + Send>,
-    ) -> Result<compute::NodeInfo, AuthError> {
-        // This method is just a convenient facade for `handle_user`
-        super::handle_user(config, client, self).await
+        client: &mut PqStream<impl AsyncRead + AsyncWrite + Unpin>,
+    ) -> Result<DatabaseInfo, AuthError> {
+        use crate::config::ClientAuthMethod::*;
+        use crate::config::RouterConfig::*;
+        match &config.router_config {
+            Static { host, port } => super::handle_static(host.clone(), *port, client, self).await,
+            Dynamic(Mixed) => {
+                if self.user.ends_with("@zenith") {
+                    super::handle_existing_user(config, client, self).await
+                } else {
+                    super::handle_new_user(config, client).await
+                }
+            }
+            Dynamic(Password) => super::handle_existing_user(config, client, self).await,
+            Dynamic(Link) => super::handle_new_user(config, client).await,
+        }
     }
 }

proxy/src/auth/flow.rs

@@ -27,6 +27,19 @@ impl AuthMethod for Scram<'_> {
     }
 }
 
+/// Use password-based auth in [`AuthFlow`].
+pub struct Md5(
+    /// Salt for client.
+    pub [u8; 4],
+);
+
+impl AuthMethod for Md5 {
+    #[inline(always)]
+    fn first_message(&self) -> BeMessage<'_> {
+        Be::AuthenticationMD5Password(self.0)
+    }
+}
+
 /// This wrapper for [`PqStream`] performs client authentication.
 #[must_use]
 pub struct AuthFlow<'a, Stream, State> {
@@ -57,10 +70,19 @@ impl<'a, S: AsyncWrite + Unpin> AuthFlow<'a, S, Begin> {
     }
 }
 
+/// Stream wrapper for handling simple MD5 password auth.
+impl<S: AsyncRead + AsyncWrite + Unpin> AuthFlow<'_, S, Md5> {
+    /// Perform user authentication. Raise an error in case authentication failed.
+    #[allow(unused)]
+    pub async fn authenticate(self) -> Result<(), AuthError> {
+        unimplemented!("MD5 auth flow is yet to be implemented");
+    }
+}
+
 /// Stream wrapper for handling [SCRAM](crate::scram) auth.
 impl<S: AsyncRead + AsyncWrite + Unpin> AuthFlow<'_, S, Scram<'_>> {
     /// Perform user authentication. Raise an error in case authentication failed.
-    pub async fn authenticate(self) -> Result<scram::ScramKey, AuthError> {
+    pub async fn authenticate(self) -> Result<(), AuthError> {
         // Initial client message contains the chosen auth method's name.
         let msg = self.stream.read_password_message().await?;
         let sasl = sasl::FirstMessage::parse(&msg).ok_or(AuthErrorImpl::MalformedPassword)?;
@@ -71,10 +93,10 @@ impl<S: AsyncRead + AsyncWrite + Unpin> AuthFlow<'_, S, Scram<'_>> {
         }
 
         let secret = self.state.0;
-        let key = sasl::SaslStream::new(self.stream, sasl.message)
+        sasl::SaslStream::new(self.stream, sasl.message)
             .authenticate(scram::Exchange::new(secret, rand::random, None))
             .await?;
 
-        Ok(key)
+        Ok(())
     }
 }

proxy/src/auth_backend.rs

@@ -1,31 +0,0 @@
-pub mod console;
-pub mod legacy_console;
-pub mod link;
-pub mod postgres;
-
-pub use legacy_console::{AuthError, AuthErrorImpl};
-
-use crate::mgmt;
-use crate::waiters::{self, Waiter, Waiters};
-use lazy_static::lazy_static;
-
-lazy_static! {
-    static ref CPLANE_WAITERS: Waiters<mgmt::ComputeReady> = Default::default();
-}
-
-/// Give caller an opportunity to wait for the cloud's reply.
-pub async fn with_waiter<R, T, E>(
-    psql_session_id: impl Into<String>,
-    action: impl FnOnce(Waiter<'static, mgmt::ComputeReady>) -> R,
-) -> Result<T, E>
-where
-    R: std::future::Future<Output = Result<T, E>>,
-    E: From<waiters::RegisterError>,
-{
-    let waiter = CPLANE_WAITERS.register(psql_session_id.into())?;
-    action(waiter).await
-}
-
-pub fn notify(psql_session_id: &str, msg: mgmt::ComputeReady) -> Result<(), waiters::NotifyError> {
-    CPLANE_WAITERS.notify(psql_session_id, msg)
-}

proxy/src/auth_backend/console.rs

@@ -1,243 +0,0 @@
-//! Declaration of Cloud API V2.
-
-use crate::{
-    auth::{self, AuthFlow},
-    compute, scram,
-};
-use serde::{Deserialize, Serialize};
-use thiserror::Error;
-
-use crate::auth::ClientCredentials;
-use crate::stream::PqStream;
-
-use tokio::io::{AsyncRead, AsyncWrite};
-use utils::pq_proto::{BeMessage as Be, BeParameterStatusMessage};
-
-#[derive(Debug, Error)]
-pub enum ConsoleAuthError {
-    // We shouldn't include the actual secret here.
-    #[error("Bad authentication secret")]
-    BadSecret,
-
-    #[error("Bad client credentials: {0:?}")]
-    BadCredentials(crate::auth::ClientCredentials),
-
-    #[error("SNI info is missing, please upgrade the postgres client library")]
-    SniMissing,
-
-    #[error("Unexpected SNI content")]
-    SniWrong,
-
-    #[error(transparent)]
-    BadUrl(#[from] url::ParseError),
-
-    #[error(transparent)]
-    Io(#[from] std::io::Error),
-
-    /// HTTP status (other than 200) returned by the console.
-    #[error("Console responded with an HTTP status: {0}")]
-    HttpStatus(reqwest::StatusCode),
-
-    #[error(transparent)]
-    Transport(#[from] reqwest::Error),
-
-    #[error("Console responded with a malformed JSON: '{0}'")]
-    MalformedResponse(#[from] serde_json::Error),
-
-    #[error("Console responded with a malformed compute address: '{0}'")]
-    MalformedComputeAddress(String),
-}
-
-#[derive(Serialize, Deserialize, Debug)]
-struct GetRoleSecretResponse {
-    role_secret: String,
-}
-
-#[derive(Serialize, Deserialize, Debug)]
-struct GetWakeComputeResponse {
-    address: String,
-}
-
-/// Auth secret which is managed by the cloud.
-pub enum AuthInfo {
-    /// Md5 hash of user's password.
-    Md5([u8; 16]),
-    /// [SCRAM](crate::scram) authentication info.
-    Scram(scram::ServerSecret),
-}
-
-/// Compute node connection params provided by the cloud.
-/// Note how it implements serde traits, since we receive it over the wire.
-#[derive(Serialize, Deserialize, Default)]
-pub struct DatabaseInfo {
-    pub host: String,
-    pub port: u16,
-    pub dbname: String,
-    pub user: String,
-
-    /// [Cloud API V1](super::legacy) returns cleartext password,
-    /// but [Cloud API V2](super::api) implements [SCRAM](crate::scram)
-    /// authentication, so we can leverage this method and cope without password.
-    pub password: Option<String>,
-}
-
-// Manually implement debug to omit personal and sensitive info.
-impl std::fmt::Debug for DatabaseInfo {
-    fn fmt(&self, fmt: &mut std::fmt::Formatter) -> std::fmt::Result {
-        fmt.debug_struct("DatabaseInfo")
-            .field("host", &self.host)
-            .field("port", &self.port)
-            .finish()
-    }
-}
-
-impl From<DatabaseInfo> for tokio_postgres::Config {
-    fn from(db_info: DatabaseInfo) -> Self {
-        let mut config = tokio_postgres::Config::new();
-
-        config
-            .host(&db_info.host)
-            .port(db_info.port)
-            .dbname(&db_info.dbname)
-            .user(&db_info.user);
-
-        if let Some(password) = db_info.password {
-            config.password(password);
-        }
-
-        config
-    }
-}
-
-async fn get_auth_info(
-    auth_endpoint: &str,
-    user: &str,
-    cluster: &str,
-) -> Result<AuthInfo, ConsoleAuthError> {
-    let mut url = reqwest::Url::parse(&format!("{auth_endpoint}/proxy_get_role_secret"))?;
-
-    url.query_pairs_mut()
-        .append_pair("cluster", cluster)
-        .append_pair("role", user);
-
-    // TODO: use a proper logger
-    println!("cplane request: {}", url);
-
-    let resp = reqwest::get(url).await?;
-    if !resp.status().is_success() {
-        return Err(ConsoleAuthError::HttpStatus(resp.status()));
-    }
-
-    let response: GetRoleSecretResponse = serde_json::from_str(resp.text().await?.as_str())?;
-
-    scram::ServerSecret::parse(response.role_secret.as_str())
-        .map(AuthInfo::Scram)
-        .ok_or(ConsoleAuthError::BadSecret)
-}
-
-/// Wake up the compute node and return the corresponding connection info.
-async fn wake_compute(
-    auth_endpoint: &str,
-    cluster: &str,
-) -> Result<(String, u16), ConsoleAuthError> {
-    let mut url = reqwest::Url::parse(&format!("{auth_endpoint}/proxy_wake_compute"))?;
-    url.query_pairs_mut().append_pair("cluster", cluster);
-
-    // TODO: use a proper logger
-    println!("cplane request: {}", url);
-
-    let resp = reqwest::get(url).await?;
-    if !resp.status().is_success() {
-        return Err(ConsoleAuthError::HttpStatus(resp.status()));
-    }
-
-    let response: GetWakeComputeResponse = serde_json::from_str(resp.text().await?.as_str())?;
-    let (host, port) = response
-        .address
-        .split_once(':')
-        .ok_or_else(|| ConsoleAuthError::MalformedComputeAddress(response.address.clone()))?;
-    let port: u16 = port
-        .parse()
-        .map_err(|_| ConsoleAuthError::MalformedComputeAddress(response.address.clone()))?;
-
-    Ok((host.to_string(), port))
-}
-
-pub async fn handle_user(
-    auth_endpoint: &str,
-    client: &mut PqStream<impl AsyncRead + AsyncWrite + Unpin>,
-    creds: &ClientCredentials,
-) -> Result<compute::NodeInfo, crate::auth::AuthError> {
-    // Determine cluster name from SNI.
-    let cluster = creds
-        .sni_data
-        .as_ref()
-        .ok_or(ConsoleAuthError::SniMissing)?
-        .split_once('.')
-        .ok_or(ConsoleAuthError::SniWrong)?
-        .0;
-
-    let user = creds.user.as_str();
-
-    // Step 1: get the auth secret
-    let auth_info = get_auth_info(auth_endpoint, user, cluster).await?;
-
-    let flow = AuthFlow::new(client);
-    let scram_keys = match auth_info {
-        AuthInfo::Md5(_) => {
-            // TODO: decide if we should support MD5 in api v2
-            return Err(crate::auth::AuthErrorImpl::auth_failed("MD5 is not supported").into());
-        }
-        AuthInfo::Scram(secret) => {
-            let scram = auth::Scram(&secret);
-            Some(compute::ScramKeys {
-                client_key: flow.begin(scram).await?.authenticate().await?.as_bytes(),
-                server_key: secret.server_key.as_bytes(),
-            })
-        }
-    };
-
-    client
-        .write_message_noflush(&Be::AuthenticationOk)?
-        .write_message_noflush(&BeParameterStatusMessage::encoding())?;
-
-    // Step 2: wake compute
-    let (host, port) = wake_compute(auth_endpoint, cluster).await?;
-
-    Ok(compute::NodeInfo {
-        db_info: DatabaseInfo {
-            host,
-            port,
-            dbname: creds.dbname.clone(),
-            user: creds.user.clone(),
-            password: None,
-        },
-        scram_keys,
-    })
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use serde_json::json;
-
-    #[test]
-    fn parse_db_info() -> anyhow::Result<()> {
-        let _: DatabaseInfo = serde_json::from_value(json!({
-            "host": "localhost",
-            "port": 5432,
-            "dbname": "postgres",
-            "user": "john_doe",
-            "password": "password",
-        }))?;
-
-        let _: DatabaseInfo = serde_json::from_value(json!({
-            "host": "localhost",
-            "port": 5432,
-            "dbname": "postgres",
-            "user": "john_doe",
-        }))?;
-
-        Ok(())
-    }
-}

proxy/src/auth_backend/legacy_console.rs

@@ -1,206 +0,0 @@
-//! Cloud API V1.
-
-use super::console::DatabaseInfo;
-
-use crate::auth::ClientCredentials;
-use crate::stream::PqStream;
-
-use crate::{compute, waiters};
-use serde::{Deserialize, Serialize};
-
-use tokio::io::{AsyncRead, AsyncWrite};
-use utils::pq_proto::{BeMessage as Be, BeParameterStatusMessage};
-
-use thiserror::Error;
-
-use crate::error::UserFacingError;
-
-#[derive(Debug, Error)]
-pub enum AuthErrorImpl {
-    /// Authentication error reported by the console.
-    #[error("Authentication failed: {0}")]
-    AuthFailed(String),
-
-    /// HTTP status (other than 200) returned by the console.
-    #[error("Console responded with an HTTP status: {0}")]
-    HttpStatus(reqwest::StatusCode),
-
-    #[error("Console responded with a malformed JSON: {0}")]
-    MalformedResponse(#[from] serde_json::Error),
-
-    #[error(transparent)]
-    Transport(#[from] reqwest::Error),
-
-    #[error(transparent)]
-    WaiterRegister(#[from] waiters::RegisterError),
-
-    #[error(transparent)]
-    WaiterWait(#[from] waiters::WaitError),
-}
-
-#[derive(Debug, Error)]
-#[error(transparent)]
-pub struct AuthError(Box<AuthErrorImpl>);
-
-impl AuthError {
-    /// Smart constructor for authentication error reported by `mgmt`.
-    pub fn auth_failed(msg: impl Into<String>) -> Self {
-        AuthError(Box::new(AuthErrorImpl::AuthFailed(msg.into())))
-    }
-}
-
-impl<T> From<T> for AuthError
-where
-    AuthErrorImpl: From<T>,
-{
-    fn from(e: T) -> Self {
-        AuthError(Box::new(e.into()))
-    }
-}
-
-impl UserFacingError for AuthError {
-    fn to_string_client(&self) -> String {
-        use AuthErrorImpl::*;
-        match self.0.as_ref() {
-            AuthFailed(_) | HttpStatus(_) => self.to_string(),
-            _ => "Internal error".to_string(),
-        }
-    }
-}
-
-// NOTE: the order of constructors is important.
-// https://serde.rs/enum-representations.html#untagged
-#[derive(Serialize, Deserialize, Debug)]
-#[serde(untagged)]
-enum ProxyAuthResponse {
-    Ready { conn_info: DatabaseInfo },
-    Error { error: String },
-    NotReady { ready: bool }, // TODO: get rid of `ready`
-}
-
-async fn authenticate_proxy_client(
-    auth_endpoint: &reqwest::Url,
-    creds: &ClientCredentials,
-    md5_response: &str,
-    salt: &[u8; 4],
-    psql_session_id: &str,
-) -> Result<DatabaseInfo, AuthError> {
-    let mut url = auth_endpoint.clone();
-    url.query_pairs_mut()
-        .append_pair("login", &creds.user)
-        .append_pair("database", &creds.dbname)
-        .append_pair("md5response", md5_response)
-        .append_pair("salt", &hex::encode(salt))
-        .append_pair("psql_session_id", psql_session_id);
-
-    super::with_waiter(psql_session_id, |waiter| async {
-        println!("cloud request: {}", url);
-        // TODO: leverage `reqwest::Client` to reuse connections
-        let resp = reqwest::get(url).await?;
-        if !resp.status().is_success() {
-            return Err(AuthErrorImpl::HttpStatus(resp.status()).into());
-        }
-
-        let auth_info: ProxyAuthResponse = serde_json::from_str(resp.text().await?.as_str())?;
-        println!("got auth info: #{:?}", auth_info);
-
-        use ProxyAuthResponse::*;
-        let db_info = match auth_info {
-            Ready { conn_info } => conn_info,
-            Error { error } => return Err(AuthErrorImpl::AuthFailed(error).into()),
-            NotReady { .. } => waiter.await?.map_err(AuthErrorImpl::AuthFailed)?,
-        };
-
-        Ok(db_info)
-    })
-    .await
-}
-
-async fn handle_existing_user(
-    auth_endpoint: &reqwest::Url,
-    client: &mut PqStream<impl AsyncRead + AsyncWrite + Unpin + Send>,
-    creds: &ClientCredentials,
-) -> Result<crate::compute::NodeInfo, crate::auth::AuthError> {
-    let psql_session_id = super::link::new_psql_session_id();
-    let md5_salt = rand::random();
-
-    client
-        .write_message(&Be::AuthenticationMD5Password(md5_salt))
-        .await?;
-
-    // Read client's password hash
-    let msg = client.read_password_message().await?;
-    let md5_response = parse_password(&msg).ok_or(crate::auth::AuthErrorImpl::MalformedPassword)?;
-
-    let db_info = authenticate_proxy_client(
-        auth_endpoint,
-        creds,
-        md5_response,
-        &md5_salt,
-        &psql_session_id,
-    )
-    .await?;
-
-    client
-        .write_message_noflush(&Be::AuthenticationOk)?
-        .write_message_noflush(&BeParameterStatusMessage::encoding())?;
-
-    Ok(compute::NodeInfo {
-        db_info,
-        scram_keys: None,
-    })
-}
-
-pub async fn handle_user(
-    auth_endpoint: &reqwest::Url,
-    auth_link_uri: &reqwest::Url,
-    client: &mut PqStream<impl AsyncRead + AsyncWrite + Unpin + Send>,
-    creds: &ClientCredentials,
-) -> Result<crate::compute::NodeInfo, crate::auth::AuthError> {
-    if creds.is_existing_user() {
-        handle_existing_user(auth_endpoint, client, creds).await
-    } else {
-        super::link::handle_user(auth_link_uri.as_ref(), client).await
-    }
-}
-
-fn parse_password(bytes: &[u8]) -> Option<&str> {
-    std::str::from_utf8(bytes).ok()?.strip_suffix('\0')
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use serde_json::json;
-
-    #[test]
-    fn test_proxy_auth_response() {
-        // Ready
-        let auth: ProxyAuthResponse = serde_json::from_value(json!({
-            "ready": true,
-            "conn_info": DatabaseInfo::default(),
-        }))
-        .unwrap();
-        assert!(matches!(
-            auth,
-            ProxyAuthResponse::Ready {
-                conn_info: DatabaseInfo { .. }
-            }
-        ));
-
-        // Error
-        let auth: ProxyAuthResponse = serde_json::from_value(json!({
-            "ready": false,
-            "error": "too bad, so sad",
-        }))
-        .unwrap();
-        assert!(matches!(auth, ProxyAuthResponse::Error { .. }));
-
-        // NotReady
-        let auth: ProxyAuthResponse = serde_json::from_value(json!({
-            "ready": false,
-        }))
-        .unwrap();
-        assert!(matches!(auth, ProxyAuthResponse::NotReady { .. }));
-    }
-}

proxy/src/auth_backend/link.rs

@@ -1,52 +0,0 @@
-use crate::{compute, stream::PqStream};
-use tokio::io::{AsyncRead, AsyncWrite};
-use utils::pq_proto::{BeMessage as Be, BeParameterStatusMessage};
-
-fn hello_message(redirect_uri: &str, session_id: &str) -> String {
-    format!(
-        concat![
-            "☀️  Welcome to Neon!\n",
-            "To proceed with database creation, open the following link:\n\n",
-            "    {redirect_uri}{session_id}\n\n",
-            "It needs to be done once and we will send you '.pgpass' file,\n",
-            "which will allow you to access or create ",
-            "databases without opening your web browser."
-        ],
-        redirect_uri = redirect_uri,
-        session_id = session_id,
-    )
-}
-
-pub fn new_psql_session_id() -> String {
-    hex::encode(rand::random::<[u8; 8]>())
-}
-
-pub async fn handle_user(
-    redirect_uri: &str,
-    client: &mut PqStream<impl AsyncRead + AsyncWrite + Unpin>,
-) -> Result<compute::NodeInfo, crate::auth::AuthError> {
-    let psql_session_id = new_psql_session_id();
-    let greeting = hello_message(redirect_uri, &psql_session_id);
-
-    let db_info = crate::auth_backend::with_waiter(psql_session_id, |waiter| async {
-        // Give user a URL to spawn a new database
-        client
-            .write_message_noflush(&Be::AuthenticationOk)?
-            .write_message_noflush(&BeParameterStatusMessage::encoding())?
-            .write_message(&Be::NoticeResponse(&greeting))
-            .await?;
-
-        // Wait for web console response (see `mgmt`)
-        waiter
-            .await?
-            .map_err(crate::auth::AuthErrorImpl::auth_failed)
-    })
-    .await?;
-
-    client.write_message_noflush(&Be::NoticeResponse("Connecting to database."))?;
-
-    Ok(compute::NodeInfo {
-        db_info,
-        scram_keys: None,
-    })
-}

proxy/src/auth_backend/postgres.rs

@@ -1,93 +0,0 @@
-//! Local mock of Cloud API V2.
-
-use super::console::{self, AuthInfo, DatabaseInfo};
-use crate::scram;
-use crate::{auth::ClientCredentials, compute};
-
-use crate::stream::PqStream;
-use tokio::io::{AsyncRead, AsyncWrite};
-use utils::pq_proto::{BeMessage as Be, BeParameterStatusMessage};
-
-async fn get_auth_info(
-    auth_endpoint: &str,
-    creds: &ClientCredentials,
-) -> Result<AuthInfo, console::ConsoleAuthError> {
-    // We wrap `tokio_postgres::Error` because we don't want to infect the
-    // method's error type with a detail that's specific to debug mode only.
-    let io_error = |e| std::io::Error::new(std::io::ErrorKind::Other, e);
-
-    // Perhaps we could persist this connection, but then we'd have to
-    // write more code for reopening it if it got closed, which doesn't
-    // seem worth it.
-    let (client, connection) = tokio_postgres::connect(auth_endpoint, tokio_postgres::NoTls)
-        .await
-        .map_err(io_error)?;
-
-    tokio::spawn(connection);
-    let query = "select rolpassword from pg_catalog.pg_authid where rolname = $1";
-    let rows = client
-        .query(query, &[&creds.user])
-        .await
-        .map_err(io_error)?;
-
-    match &rows[..] {
-        // We can't get a secret if there's no such user.
-        [] => Err(console::ConsoleAuthError::BadCredentials(creds.to_owned())),
-        // We shouldn't get more than one row anyway.
-        [row, ..] => {
-            let entry = row.try_get(0).map_err(io_error)?;
-            scram::ServerSecret::parse(entry)
-                .map(AuthInfo::Scram)
-                .or_else(|| {
-                    // It could be an md5 hash if it's not a SCRAM secret.
-                    let text = entry.strip_prefix("md5")?;
-                    Some(AuthInfo::Md5({
-                        let mut bytes = [0u8; 16];
-                        hex::decode_to_slice(text, &mut bytes).ok()?;
-                        bytes
-                    }))
-                })
-                // Putting the secret into this message is a security hazard!
-                .ok_or(console::ConsoleAuthError::BadSecret)
-        }
-    }
-}
-
-pub async fn handle_user(
-    auth_endpoint: &reqwest::Url,
-    client: &mut PqStream<impl AsyncRead + AsyncWrite + Unpin>,
-    creds: &ClientCredentials,
-) -> Result<compute::NodeInfo, crate::auth::AuthError> {
-    let auth_info = get_auth_info(auth_endpoint.as_ref(), creds).await?;
-
-    let flow = crate::auth::AuthFlow::new(client);
-    let scram_keys = match auth_info {
-        AuthInfo::Md5(_) => {
-            // TODO: decide if we should support MD5 in api v2
-            return Err(crate::auth::AuthErrorImpl::auth_failed("MD5 is not supported").into());
-        }
-        AuthInfo::Scram(secret) => {
-            let scram = crate::auth::Scram(&secret);
-            Some(compute::ScramKeys {
-                client_key: flow.begin(scram).await?.authenticate().await?.as_bytes(),
-                server_key: secret.server_key.as_bytes(),
-            })
-        }
-    };
-
-    client
-        .write_message_noflush(&Be::AuthenticationOk)?
-        .write_message_noflush(&BeParameterStatusMessage::encoding())?;
-
-    Ok(compute::NodeInfo {
-        db_info: DatabaseInfo {
-            // TODO: handle that near CLI params parsing
-            host: auth_endpoint.host_str().unwrap_or("localhost").to_owned(),
-            port: auth_endpoint.port().unwrap_or(5432),
-            dbname: creds.dbname.to_owned(),
-            user: creds.user.to_owned(),
-            password: None,
-        },
-        scram_keys,
-    })
-}

proxy/src/compute.rs

@@ -1,6 +1,6 @@
-use crate::auth_backend::console::DatabaseInfo;
 use crate::cancellation::CancelClosure;
 use crate::error::UserFacingError;
+use serde::{Deserialize, Serialize};
 use std::io;
 use std::net::SocketAddr;
 use thiserror::Error;
@@ -23,21 +23,32 @@ pub enum ConnectionError {
 
 impl UserFacingError for ConnectionError {}
 
+/// Compute node connection params.
+#[derive(Serialize, Deserialize, Default)]
+pub struct DatabaseInfo {
+    pub host: String,
+    pub port: u16,
+    pub dbname: String,
+    pub user: String,
+    pub password: Option<String>,
+}
+
+// Manually implement debug to omit personal and sensitive info
+impl std::fmt::Debug for DatabaseInfo {
+    fn fmt(&self, fmt: &mut std::fmt::Formatter) -> std::fmt::Result {
+        fmt.debug_struct("DatabaseInfo")
+            .field("host", &self.host)
+            .field("port", &self.port)
+            .finish()
+    }
+}
+
 /// PostgreSQL version as [`String`].
 pub type Version = String;
 
-/// A pair of `ClientKey` & `ServerKey` for `SCRAM-SHA-256`.
-pub type ScramKeys = tokio_postgres::config::ScramKeys<32>;
-
-/// Compute node connection params.
-pub struct NodeInfo {
-    pub db_info: DatabaseInfo,
-    pub scram_keys: Option<ScramKeys>,
-}
-
-impl NodeInfo {
+impl DatabaseInfo {
     async fn connect_raw(&self) -> io::Result<(SocketAddr, TcpStream)> {
-        let host_port = format!("{}:{}", self.db_info.host, self.db_info.port);
+        let host_port = format!("{}:{}", self.host, self.port);
         let socket = TcpStream::connect(host_port).await?;
         let socket_addr = socket.peer_addr()?;
         socket2::SockRef::from(&socket).set_keepalive(true)?;
@@ -52,13 +63,11 @@ impl NodeInfo {
             .await
             .map_err(|_| ConnectionError::FailedToConnectToCompute)?;
 
-        let mut config = tokio_postgres::Config::from(self.db_info);
-        if let Some(scram_keys) = self.scram_keys {
-            config.auth_keys(tokio_postgres::config::AuthKeys::ScramSha256(scram_keys));
-        }
-
         // TODO: establish a secure connection to the DB
-        let (client, conn) = config.connect_raw(&mut socket, NoTls).await?;
+        let (client, conn) = tokio_postgres::Config::from(self)
+            .connect_raw(&mut socket, NoTls)
+            .await?;
+
         let version = conn
             .parameter("server_version")
             .ok_or(ConnectionError::FailedToFetchPgVersion)?
@@ -69,3 +78,21 @@ impl NodeInfo {
         Ok((socket, version, cancel_closure))
     }
 }
+
+impl From<DatabaseInfo> for tokio_postgres::Config {
+    fn from(db_info: DatabaseInfo) -> Self {
+        let mut config = tokio_postgres::Config::new();
+
+        config
+            .host(&db_info.host)
+            .port(db_info.port)
+            .dbname(&db_info.dbname)
+            .user(&db_info.user);
+
+        if let Some(password) = db_info.password {
+            config.password(password);
+        }
+
+        config
+    }
+}

proxy/src/config.rs

@@ -1,47 +1,65 @@
-use anyhow::{ensure, Context};
-use std::{str::FromStr, sync::Arc};
+use anyhow::{bail, ensure, Context};
+use std::net::SocketAddr;
+use std::str::FromStr;
+use std::sync::Arc;
+
+pub type TlsConfig = Arc<rustls::ServerConfig>;
 
 #[non_exhaustive]
-pub enum AuthBackendType {
-    LegacyConsole,
-    Console,
-    Postgres,
+pub enum ClientAuthMethod {
+    Password,
     Link,
+
+    /// Use password auth only if username ends with "@zenith"
+    Mixed,
 }
 
-impl FromStr for AuthBackendType {
+pub enum RouterConfig {
+    Static { host: String, port: u16 },
+    Dynamic(ClientAuthMethod),
+}
+
+impl FromStr for ClientAuthMethod {
     type Err = anyhow::Error;
 
     fn from_str(s: &str) -> anyhow::Result<Self> {
-        println!("ClientAuthMethod::from_str: '{}'", s);
-        use AuthBackendType::*;
+        use ClientAuthMethod::*;
         match s {
-            "legacy" => Ok(LegacyConsole),
-            "console" => Ok(Console),
-            "postgres" => Ok(Postgres),
+            "password" => Ok(Password),
             "link" => Ok(Link),
-            _ => Err(anyhow::anyhow!("Invlid option for auth method")),
+            "mixed" => Ok(Mixed),
+            _ => bail!("Invalid option for router: `{}`", s),
         }
     }
 }
 
 pub struct ProxyConfig {
-    /// TLS configuration for the proxy.
-    pub tls_config: Option<TlsConfig>,
+    /// main entrypoint for users to connect to
+    pub proxy_address: SocketAddr,
 
-    pub auth_backend: AuthBackendType,
+    /// method of assigning compute nodes
+    pub router_config: RouterConfig,
 
+    /// internally used for status and prometheus metrics
+    pub http_address: SocketAddr,
+
+    /// management endpoint. Upon user account creation control plane
+    /// will notify us here, so that we can 'unfreeze' user session.
+    /// TODO It uses postgres protocol over TCP but should be migrated to http.
+    pub mgmt_address: SocketAddr,
+
+    /// send unauthenticated users to this URI
+    pub redirect_uri: String,
+
+    /// control plane address where we would check auth.
     pub auth_endpoint: reqwest::Url,
 
-    pub auth_link_uri: reqwest::Url,
+    pub tls_config: Option<TlsConfig>,
 }
 
-pub type TlsConfig = Arc<rustls::ServerConfig>;
-
-/// Configure TLS for the main endpoint.
-pub fn configure_tls(key_path: &str, cert_path: &str) -> anyhow::Result<TlsConfig> {
+pub fn configure_ssl(key_path: &str, cert_path: &str) -> anyhow::Result<TlsConfig> {
     let key = {
-        let key_bytes = std::fs::read(key_path).context("TLS key file")?;
+        let key_bytes = std::fs::read(key_path).context("SSL key file")?;
         let mut keys = rustls_pemfile::pkcs8_private_keys(&mut &key_bytes[..])
             .context("couldn't read TLS keys")?;
 
@@ -50,7 +68,7 @@ pub fn configure_tls(key_path: &str, cert_path: &str) -> anyhow::Result<TlsConfi
     };
 
     let cert_chain = {
-        let cert_chain_bytes = std::fs::read(cert_path).context("TLS cert file")?;
+        let cert_chain_bytes = std::fs::read(cert_path).context("SSL cert file")?;
         rustls_pemfile::certs(&mut &cert_chain_bytes[..])
             .context("couldn't read TLS certificate chain")?
             .into_iter()

proxy/src/cplane_api.rs

@@ -0,0 +1,183 @@
+use crate::auth::ClientCredentials;
+use crate::compute::DatabaseInfo;
+use crate::error::UserFacingError;
+use crate::mgmt;
+use crate::waiters::{self, Waiter, Waiters};
+use lazy_static::lazy_static;
+use serde::{Deserialize, Serialize};
+use thiserror::Error;
+
+lazy_static! {
+    static ref CPLANE_WAITERS: Waiters<mgmt::ComputeReady> = Default::default();
+}
+
+/// Give caller an opportunity to wait for cplane's reply.
+pub async fn with_waiter<R, T, E>(
+    psql_session_id: impl Into<String>,
+    action: impl FnOnce(Waiter<'static, mgmt::ComputeReady>) -> R,
+) -> Result<T, E>
+where
+    R: std::future::Future<Output = Result<T, E>>,
+    E: From<waiters::RegisterError>,
+{
+    let waiter = CPLANE_WAITERS.register(psql_session_id.into())?;
+    action(waiter).await
+}
+
+pub fn notify(
+    psql_session_id: &str,
+    msg: Result<DatabaseInfo, String>,
+) -> Result<(), waiters::NotifyError> {
+    CPLANE_WAITERS.notify(psql_session_id, msg)
+}
+
+/// Zenith console API wrapper.
+pub struct CPlaneApi {
+    auth_endpoint: reqwest::Url,
+}
+
+impl CPlaneApi {
+    pub fn new(auth_endpoint: reqwest::Url) -> Self {
+        Self { auth_endpoint }
+    }
+}
+
+#[derive(Debug, Error)]
+pub enum AuthErrorImpl {
+    /// Authentication error reported by the console.
+    #[error("Authentication failed: {0}")]
+    AuthFailed(String),
+
+    /// HTTP status (other than 200) returned by the console.
+    #[error("Console responded with an HTTP status: {0}")]
+    HttpStatus(reqwest::StatusCode),
+
+    #[error("Console responded with a malformed JSON: {0}")]
+    MalformedResponse(#[from] serde_json::Error),
+
+    #[error(transparent)]
+    Transport(#[from] reqwest::Error),
+
+    #[error(transparent)]
+    WaiterRegister(#[from] waiters::RegisterError),
+
+    #[error(transparent)]
+    WaiterWait(#[from] waiters::WaitError),
+}
+
+#[derive(Debug, Error)]
+#[error(transparent)]
+pub struct AuthError(Box<AuthErrorImpl>);
+
+impl AuthError {
+    /// Smart constructor for authentication error reported by `mgmt`.
+    pub fn auth_failed(msg: impl Into<String>) -> Self {
+        AuthError(Box::new(AuthErrorImpl::AuthFailed(msg.into())))
+    }
+}
+
+impl<T> From<T> for AuthError
+where
+    AuthErrorImpl: From<T>,
+{
+    fn from(e: T) -> Self {
+        AuthError(Box::new(e.into()))
+    }
+}
+
+impl UserFacingError for AuthError {
+    fn to_string_client(&self) -> String {
+        use AuthErrorImpl::*;
+        match self.0.as_ref() {
+            AuthFailed(_) | HttpStatus(_) => self.to_string(),
+            _ => "Internal error".to_string(),
+        }
+    }
+}
+
+impl CPlaneApi {
+    pub async fn authenticate_proxy_client(
+        &self,
+        creds: ClientCredentials,
+        md5_response: &str,
+        salt: &[u8; 4],
+        psql_session_id: &str,
+    ) -> Result<DatabaseInfo, AuthError> {
+        let mut url = self.auth_endpoint.clone();
+        url.query_pairs_mut()
+            .append_pair("login", &creds.user)
+            .append_pair("database", &creds.dbname)
+            .append_pair("md5response", md5_response)
+            .append_pair("salt", &hex::encode(salt))
+            .append_pair("psql_session_id", psql_session_id);
+
+        with_waiter(psql_session_id, |waiter| async {
+            println!("cplane request: {}", url);
+            // TODO: leverage `reqwest::Client` to reuse connections
+            let resp = reqwest::get(url).await?;
+            if !resp.status().is_success() {
+                return Err(AuthErrorImpl::HttpStatus(resp.status()).into());
+            }
+
+            let auth_info: ProxyAuthResponse = serde_json::from_str(resp.text().await?.as_str())?;
+            println!("got auth info: #{:?}", auth_info);
+
+            use ProxyAuthResponse::*;
+            let db_info = match auth_info {
+                Ready { conn_info } => conn_info,
+                Error { error } => return Err(AuthErrorImpl::AuthFailed(error).into()),
+                NotReady { .. } => waiter.await?.map_err(AuthErrorImpl::AuthFailed)?,
+            };
+
+            Ok(db_info)
+        })
+        .await
+    }
+}
+
+// NOTE: the order of constructors is important.
+// https://serde.rs/enum-representations.html#untagged
+#[derive(Serialize, Deserialize, Debug)]
+#[serde(untagged)]
+enum ProxyAuthResponse {
+    Ready { conn_info: DatabaseInfo },
+    Error { error: String },
+    NotReady { ready: bool }, // TODO: get rid of `ready`
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    #[test]
+    fn test_proxy_auth_response() {
+        // Ready
+        let auth: ProxyAuthResponse = serde_json::from_value(json!({
+            "ready": true,
+            "conn_info": DatabaseInfo::default(),
+        }))
+        .unwrap();
+        assert!(matches!(
+            auth,
+            ProxyAuthResponse::Ready {
+                conn_info: DatabaseInfo { .. }
+            }
+        ));
+
+        // Error
+        let auth: ProxyAuthResponse = serde_json::from_value(json!({
+            "ready": false,
+            "error": "too bad, so sad",
+        }))
+        .unwrap();
+        assert!(matches!(auth, ProxyAuthResponse::Error { .. }));
+
+        // NotReady
+        let auth: ProxyAuthResponse = serde_json::from_value(json!({
+            "ready": false,
+        }))
+        .unwrap();
+        assert!(matches!(auth, ProxyAuthResponse::NotReady { .. }));
+    }
+}

proxy/src/main.rs

@@ -5,28 +5,35 @@
 //! in somewhat transparent manner (again via communication with control plane API).
 
 mod auth;
-mod auth_backend;
 mod cancellation;
 mod compute;
 mod config;
+mod cplane_api;
 mod error;
 mod http;
 mod mgmt;
-mod parse;
 mod proxy;
-mod sasl;
-mod scram;
 mod stream;
 mod waiters;
 
+// Currently SCRAM is only used in tests
+#[cfg(test)]
+mod parse;
+#[cfg(test)]
+mod sasl;
+#[cfg(test)]
+mod scram;
+
 use anyhow::{bail, Context};
 use clap::{App, Arg};
 use config::ProxyConfig;
 use futures::FutureExt;
-use std::{future::Future, net::SocketAddr};
+use std::future::Future;
 use tokio::{net::TcpListener, task::JoinError};
 use utils::GIT_VERSION;
 
+use crate::config::{ClientAuthMethod, RouterConfig};
+
 /// Flattens `Result<Result<T>>` into `Result<T>`.
 async fn flatten_err(
     f: impl Future<Output = Result<anyhow::Result<()>, JoinError>>,
@@ -37,7 +44,7 @@ async fn flatten_err(
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
     metrics::set_common_metrics_prefix("zenith_proxy");
-    let arg_matches = App::new("Neon proxy/router")
+    let arg_matches = App::new("Zenith proxy/router")
         .version(GIT_VERSION)
         .arg(
             Arg::new("proxy")
@@ -48,11 +55,18 @@ async fn main() -> anyhow::Result<()> {
                 .default_value("127.0.0.1:4432"),
         )
         .arg(
-            Arg::new("auth-backend")
-                .long("auth-backend")
+            Arg::new("auth-method")
+                .long("auth-method")
                 .takes_value(true)
-                .help("Possible values: legacy | console | postgres | link")
-                .default_value("legacy"),
+                .help("Possible values: password | link | mixed")
+                .default_value("mixed"),
+        )
+        .arg(
+            Arg::new("static-router")
+                .short('s')
+                .long("static-router")
+                .takes_value(true)
+                .help("Route all clients to host:port"),
         )
         .arg(
             Arg::new("mgmt")
@@ -75,7 +89,7 @@ async fn main() -> anyhow::Result<()> {
                 .short('u')
                 .long("uri")
                 .takes_value(true)
-                .help("redirect unauthenticated users to the given uri in case of link auth")
+                .help("redirect unauthenticated users to given uri")
                 .default_value("http://localhost:3000/psql_session/"),
         )
         .arg(
@@ -83,68 +97,77 @@ async fn main() -> anyhow::Result<()> {
                 .short('a')
                 .long("auth-endpoint")
                 .takes_value(true)
-                .help("cloud API endpoint for authenticating users")
+                .help("API endpoint for authenticating users")
                 .default_value("http://localhost:3000/authenticate_proxy_request/"),
         )
         .arg(
-            Arg::new("tls-key")
+            Arg::new("ssl-key")
                 .short('k')
-                .long("tls-key")
-                .alias("ssl-key") // backwards compatibility
+                .long("ssl-key")
                 .takes_value(true)
-                .help("path to TLS key for client postgres connections"),
+                .help("path to SSL key for client postgres connections"),
         )
         .arg(
-            Arg::new("tls-cert")
+            Arg::new("ssl-cert")
                 .short('c')
-                .long("tls-cert")
-                .alias("ssl-cert") // backwards compatibility
+                .long("ssl-cert")
                 .takes_value(true)
-                .help("path to TLS cert for client postgres connections"),
+                .help("path to SSL cert for client postgres connections"),
         )
         .get_matches();
 
     let tls_config = match (
-        arg_matches.value_of("tls-key"),
-        arg_matches.value_of("tls-cert"),
+        arg_matches.value_of("ssl-key"),
+        arg_matches.value_of("ssl-cert"),
     ) {
-        (Some(key_path), Some(cert_path)) => Some(config::configure_tls(key_path, cert_path)?),
+        (Some(key_path), Some(cert_path)) => Some(config::configure_ssl(key_path, cert_path)?),
         (None, None) => None,
-        _ => bail!("either both or neither tls-key and tls-cert must be specified"),
+        _ => bail!("either both or neither ssl-key and ssl-cert must be specified"),
     };
 
-    let proxy_address: SocketAddr = arg_matches.value_of("proxy").unwrap().parse()?;
-    let mgmt_address: SocketAddr = arg_matches.value_of("mgmt").unwrap().parse()?;
-    let http_address: SocketAddr = arg_matches.value_of("http").unwrap().parse()?;
+    let auth_method = arg_matches.value_of("auth-method").unwrap().parse()?;
+    let router_config = match arg_matches.value_of("static-router") {
+        None => RouterConfig::Dynamic(auth_method),
+        Some(addr) => {
+            if let ClientAuthMethod::Password = auth_method {
+                let (host, port) = addr.split_once(':').unwrap();
+                RouterConfig::Static {
+                    host: host.to_string(),
+                    port: port.parse().unwrap(),
+                }
+            } else {
+                bail!("static-router requires --auth-method password")
+            }
+        }
+    };
 
     let config: &ProxyConfig = Box::leak(Box::new(ProxyConfig {
-        tls_config,
-        auth_backend: arg_matches.value_of("auth-backend").unwrap().parse()?,
+        router_config,
+        proxy_address: arg_matches.value_of("proxy").unwrap().parse()?,
+        mgmt_address: arg_matches.value_of("mgmt").unwrap().parse()?,
+        http_address: arg_matches.value_of("http").unwrap().parse()?,
+        redirect_uri: arg_matches.value_of("uri").unwrap().parse()?,
         auth_endpoint: arg_matches.value_of("auth-endpoint").unwrap().parse()?,
-        auth_link_uri: arg_matches.value_of("uri").unwrap().parse()?,
+        tls_config,
     }));
 
     println!("Version: {}", GIT_VERSION);
 
     // Check that we can bind to address before further initialization
-    println!("Starting http on {}", http_address);
-    let http_listener = TcpListener::bind(http_address).await?.into_std()?;
+    println!("Starting http on {}", config.http_address);
+    let http_listener = TcpListener::bind(config.http_address).await?.into_std()?;
 
-    println!("Starting mgmt on {}", mgmt_address);
-    let mgmt_listener = TcpListener::bind(mgmt_address).await?.into_std()?;
+    println!("Starting mgmt on {}", config.mgmt_address);
+    let mgmt_listener = TcpListener::bind(config.mgmt_address).await?.into_std()?;
 
-    println!("Starting proxy on {}", proxy_address);
-    let proxy_listener = TcpListener::bind(proxy_address).await?;
+    println!("Starting proxy on {}", config.proxy_address);
+    let proxy_listener = TcpListener::bind(config.proxy_address).await?;
 
-    let tasks = [
-        tokio::spawn(http::thread_main(http_listener)),
-        tokio::spawn(proxy::thread_main(config, proxy_listener)),
-        tokio::task::spawn_blocking(move || mgmt::thread_main(mgmt_listener)),
-    ]
-    .map(flatten_err);
+    let http = tokio::spawn(http::thread_main(http_listener));
+    let proxy = tokio::spawn(proxy::thread_main(config, proxy_listener));
+    let mgmt = tokio::task::spawn_blocking(move || mgmt::thread_main(mgmt_listener));
 
-    // This will block until all tasks have completed.
-    // Furthermore, the first one to fail will cancel the rest.
+    let tasks = [flatten_err(http), flatten_err(proxy), flatten_err(mgmt)];
     let _: Vec<()> = futures::future::try_join_all(tasks).await?;
 
     Ok(())

proxy/src/mgmt.rs

@@ -1,4 +1,4 @@
-use crate::auth_backend;
+use crate::{compute::DatabaseInfo, cplane_api};
 use anyhow::Context;
 use serde::Deserialize;
 use std::{
@@ -10,8 +10,6 @@ use utils::{
     pq_proto::{BeMessage, SINGLE_COL_ROWDESC},
 };
 
-/// TODO: move all of that to auth-backend/link.rs when we ditch legacy-console backend
-
 ///
 /// Main proxy listener loop.
 ///
@@ -77,12 +75,12 @@ struct PsqlSessionResponse {
 
 #[derive(Deserialize)]
 enum PsqlSessionResult {
-    Success(auth_backend::console::DatabaseInfo),
+    Success(DatabaseInfo),
     Failure(String),
 }
 
 /// A message received by `mgmt` when a compute node is ready.
-pub type ComputeReady = Result<auth_backend::console::DatabaseInfo, String>;
+pub type ComputeReady = Result<DatabaseInfo, String>;
 
 impl PsqlSessionResult {
     fn into_compute_ready(self) -> ComputeReady {
@@ -113,7 +111,7 @@ fn try_process_query(pgb: &mut PostgresBackend, query_string: &str) -> anyhow::R
 
     let resp: PsqlSessionResponse = serde_json::from_str(query_string)?;
 
-    match auth_backend::notify(&resp.session_id, resp.result.into_compute_ready()) {
+    match cplane_api::notify(&resp.session_id, resp.result.into_compute_ready()) {
         Ok(()) => {
             pgb.write_message_noflush(&SINGLE_COL_ROWDESC)?
                 .write_message_noflush(&BeMessage::DataRow(&[Some(b"ok")]))?

proxy/src/proxy.rs

@@ -73,7 +73,7 @@ pub async fn thread_main(
 async fn handle_client(
     config: &ProxyConfig,
     cancel_map: &CancelMap,
-    stream: impl AsyncRead + AsyncWrite + Unpin + Send,
+    stream: impl AsyncRead + AsyncWrite + Unpin,
 ) -> anyhow::Result<()> {
     // The `closed` counter will increase when this future is destroyed.
     NUM_CONNECTIONS_ACCEPTED_COUNTER.inc();
@@ -144,15 +144,10 @@ async fn handshake<S: AsyncRead + AsyncWrite + Unpin>(
                 }
 
                 // Here and forth: `or_else` demands that we use a future here
-                let mut creds: auth::ClientCredentials = async { params.try_into() }
+                let creds = async { params.try_into() }
                     .or_else(|e| stream.throw_error(e))
                     .await?;
 
-                // Set SNI info when available
-                if let Stream::Tls { tls } = stream.get_ref() {
-                    creds.sni_data = tls.get_ref().1.sni_hostname().map(|s| s.to_owned());
-                }
-
                 break Ok(Some((stream, creds)));
             }
             CancelRequest(cancel_key_data) => {
@@ -179,7 +174,7 @@ impl<S> Client<S> {
     }
 }
 
-impl<S: AsyncRead + AsyncWrite + Unpin + Send> Client<S> {
+impl<S: AsyncRead + AsyncWrite + Unpin> Client<S> {
     /// Let the client authenticate and connect to the designated compute node.
     async fn connect_to_db(
         self,
@@ -190,10 +185,10 @@ impl<S: AsyncRead + AsyncWrite + Unpin + Send> Client<S> {
 
         // Authenticate and connect to a compute node.
         let auth = creds.authenticate(config, &mut stream).await;
-        let node = async { auth }.or_else(|e| stream.throw_error(e)).await?;
+        let db_info = async { auth }.or_else(|e| stream.throw_error(e)).await?;
 
         let (db, version, cancel_closure) =
-            node.connect().or_else(|e| stream.throw_error(e)).await?;
+            db_info.connect().or_else(|e| stream.throw_error(e)).await?;
         let cancel_key_data = session.enable_cancellation(cancel_closure);
 
         stream

proxy/src/sasl.rs

@@ -39,20 +39,9 @@ pub enum Error {
 /// A convenient result type for SASL exchange.
 pub type Result<T> = std::result::Result<T, Error>;
 
-/// A result of one SASL exchange.
-pub enum Step<T, R> {
-    /// We should continue exchanging messages.
-    Continue(T),
-    /// The client has been authenticated successfully.
-    Authenticated(R),
-}
-
 /// Every SASL mechanism (e.g. [SCRAM](crate::scram)) is expected to implement this trait.
 pub trait Mechanism: Sized {
-    /// What's produced as a result of successful authentication.
-    type Output;
-
     /// Produce a server challenge to be sent to the client.
     /// This is how this method is called in PostgreSQL (`libpq/sasl.h`).
-    fn exchange(self, input: &str) -> Result<(Step<Self, Self::Output>, String)>;
+    fn exchange(self, input: &str) -> Result<(Option<Self>, String)>;
 }

proxy/src/sasl/messages.rs

@@ -49,7 +49,6 @@ impl<'a> ServerMessage<&'a str> {
         })
     }
 }
-
 #[cfg(test)]
 mod tests {
     use super::*;

proxy/src/sasl/stream.rs

@@ -51,23 +51,18 @@ impl<S: AsyncWrite + Unpin> SaslStream<'_, S> {
 impl<S: AsyncRead + AsyncWrite + Unpin> SaslStream<'_, S> {
     /// Perform SASL message exchange according to the underlying algorithm
     /// until user is either authenticated or denied access.
-    pub async fn authenticate<M: Mechanism>(
-        mut self,
-        mut mechanism: M,
-    ) -> super::Result<M::Output> {
+    pub async fn authenticate(mut self, mut mechanism: impl Mechanism) -> super::Result<()> {
         loop {
             let input = self.recv().await?;
             let (moved, reply) = mechanism.exchange(input)?;
-
-            use super::Step::*;
             match moved {
-                Continue(moved) => {
+                Some(moved) => {
                     self.send(&ServerMessage::Continue(&reply)).await?;
                     mechanism = moved;
                 }
-                Authenticated(result) => {
+                None => {
                     self.send(&ServerMessage::Final(&reply)).await?;
-                    return Ok(result);
+                    return Ok(());
                 }
             }
         }

proxy/src/scram.rs

@@ -9,16 +9,14 @@
 mod exchange;
 mod key;
 mod messages;
+mod password;
 mod secret;
 mod signature;
 
-#[cfg(test)]
-mod password;
+pub use secret::*;
 
 pub use exchange::Exchange;
-pub use key::ScramKey;
 pub use secret::ServerSecret;
-pub use secret::*;
 
 use hmac::{Hmac, Mac};
 use sha2::{Digest, Sha256};

proxy/src/scram/exchange.rs

@@ -8,6 +8,7 @@ use super::signature::SignatureBuilder;
 use crate::sasl::{self, ChannelBinding, Error as SaslError};
 
 /// The only channel binding mode we currently support.
+#[derive(Debug)]
 struct TlsServerEndPoint;
 
 impl std::fmt::Display for TlsServerEndPoint {
@@ -27,6 +28,7 @@ impl std::str::FromStr for TlsServerEndPoint {
     }
 }
 
+#[derive(Debug)]
 enum ExchangeState {
     /// Waiting for [`ClientFirstMessage`].
     Initial,
@@ -39,6 +41,7 @@ enum ExchangeState {
 }
 
 /// Server's side of SCRAM auth algorithm.
+#[derive(Debug)]
 pub struct Exchange<'a> {
     state: ExchangeState,
     secret: &'a ServerSecret,
@@ -62,10 +65,8 @@ impl<'a> Exchange<'a> {
 }
 
 impl sasl::Mechanism for Exchange<'_> {
-    type Output = super::ScramKey;
-
-    fn exchange(mut self, input: &str) -> sasl::Result<(sasl::Step<Self, Self::Output>, String)> {
-        use {sasl::Step::*, ExchangeState::*};
+    fn exchange(mut self, input: &str) -> sasl::Result<(Option<Self>, String)> {
+        use ExchangeState::*;
         match &self.state {
             Initial => {
                 let client_first_message =
@@ -84,7 +85,7 @@ impl sasl::Mechanism for Exchange<'_> {
                     server_first_message,
                 };
 
-                Ok((Continue(self), msg))
+                Ok((Some(self), msg))
             }
             SaltSent {
                 cbind_flag,
@@ -126,7 +127,7 @@ impl sasl::Mechanism for Exchange<'_> {
                 let msg = client_final_message
                     .build_server_final_message(signature_builder, &self.secret.server_key);
 
-                Ok((Authenticated(client_key), msg))
+                Ok((None, msg))
             }
         }
     }

proxy/src/scram/key.rs

@@ -6,7 +6,7 @@ pub const SCRAM_KEY_LEN: usize = 32;
 /// One of the keys derived from the [password](super::password::SaltedPassword).
 /// We use the same structure for all keys, i.e.
 /// `ClientKey`, `StoredKey`, and `ServerKey`.
-#[derive(Default, PartialEq, Eq)]
+#[derive(Default, Debug, PartialEq, Eq)]
 #[repr(transparent)]
 pub struct ScramKey {
     bytes: [u8; SCRAM_KEY_LEN],
@@ -16,10 +16,6 @@ impl ScramKey {
     pub fn sha256(&self) -> Self {
         super::sha256([self.as_ref()]).into()
     }
-
-    pub fn as_bytes(&self) -> [u8; SCRAM_KEY_LEN] {
-        self.bytes
-    }
 }
 
 impl From<[u8; SCRAM_KEY_LEN]> for ScramKey {

proxy/src/scram/secret.rs

@@ -5,6 +5,7 @@ use super::key::ScramKey;
 
 /// Server secret is produced from [password](super::password::SaltedPassword)
 /// and is used throughout the authentication process.
+#[derive(Debug)]
 pub struct ServerSecret {
     /// Number of iterations for `PBKDF2` function.
     pub iterations: u32,
@@ -38,7 +39,6 @@ impl ServerSecret {
     /// To avoid revealing information to an attacker, we use a
     /// mocked server secret even if the user doesn't exist.
     /// See `auth-scram.c : mock_scram_secret` for details.
-    #[allow(dead_code)]
     pub fn mock(user: &str, nonce: &[u8; 32]) -> Self {
         // Refer to `auth-scram.c : scram_mock_salt`.
         let mocked_salt = super::sha256([user.as_bytes(), nonce]);

test_runner/batch_others/test_ancestor_branch.py

@@ -33,10 +33,6 @@ def test_ancestor_branch(zenith_env_builder: ZenithEnvBuilder):
             'compaction_target_size': '4194304',
         })
 
-    with closing(env.pageserver.connect()) as psconn:
-        with psconn.cursor(cursor_factory=psycopg2.extras.DictCursor) as pscur:
-            pscur.execute("failpoints flush-frozen=sleep(10000)")
-
     env.zenith_cli.create_timeline(f'main', tenant_id=tenant)
     pg_branch0 = env.postgres.create_start('main', tenant_id=tenant)
     branch0_cur = pg_branch0.connect().cursor()

test_runner/batch_others/test_lsn_mapping.py

@@ -1,84 +0,0 @@
-from contextlib import closing
-from datetime import timedelta, timezone, tzinfo
-import math
-from uuid import UUID
-import psycopg2.extras
-import psycopg2.errors
-from fixtures.zenith_fixtures import ZenithEnv, ZenithEnvBuilder, Postgres
-from fixtures.log_helper import log
-import time
-
-
-#
-# Test pageserver get_lsn_by_timestamp API
-#
-def test_lsn_mapping(zenith_env_builder: ZenithEnvBuilder):
-    zenith_env_builder.num_safekeepers = 1
-    env = zenith_env_builder.init_start()
-
-    new_timeline_id = env.zenith_cli.create_branch('test_lsn_mapping')
-    pgmain = env.postgres.create_start("test_lsn_mapping")
-    log.info("postgres is running on 'test_lsn_mapping' branch")
-
-    ps_conn = env.pageserver.connect()
-    ps_cur = ps_conn.cursor()
-    conn = pgmain.connect()
-    cur = conn.cursor()
-
-    # Create table, and insert rows, each in a separate transaction
-    # Disable synchronous_commit to make this initialization go faster.
-    #
-    # Each row contains current insert LSN and the current timestamp, when
-    # the row was inserted.
-    cur.execute("SET synchronous_commit=off")
-    cur.execute("CREATE TABLE foo (x integer)")
-    tbl = []
-    for i in range(1000):
-        cur.execute(f"INSERT INTO foo VALUES({i})")
-        cur.execute(f'SELECT clock_timestamp()')
-        # Get the timestamp at UTC
-        after_timestamp = cur.fetchone()[0].replace(tzinfo=None)
-        tbl.append([i, after_timestamp])
-
-    # Execute one more transaction with synchronous_commit enabled, to flush
-    # all the previous transactions
-    cur.execute("SET synchronous_commit=on")
-    cur.execute("INSERT INTO foo VALUES (-1)")
-
-    # Check edge cases: timestamp in the future
-    probe_timestamp = tbl[-1][1] + timedelta(hours=1)
-    ps_cur.execute(
-        f"get_lsn_by_timestamp {env.initial_tenant.hex} {new_timeline_id.hex} '{probe_timestamp.isoformat()}Z'"
-    )
-    result = ps_cur.fetchone()[0]
-    assert result == 'future'
-
-    # timestamp too the far history
-    probe_timestamp = tbl[0][1] - timedelta(hours=10)
-    ps_cur.execute(
-        f"get_lsn_by_timestamp {env.initial_tenant.hex} {new_timeline_id.hex} '{probe_timestamp.isoformat()}Z'"
-    )
-    result = ps_cur.fetchone()[0]
-    assert result == 'past'
-
-    # Probe a bunch of timestamps in the valid range
-    for i in range(1, len(tbl), 100):
-        probe_timestamp = tbl[i][1]
-
-        # Call get_lsn_by_timestamp to get the LSN
-        ps_cur.execute(
-            f"get_lsn_by_timestamp {env.initial_tenant.hex} {new_timeline_id.hex} '{probe_timestamp.isoformat()}Z'"
-        )
-        lsn = ps_cur.fetchone()[0]
-
-        # Launch a new read-only node at that LSN, and check that only the rows
-        # that were supposed to be committed at that point in time are visible.
-        pg_here = env.postgres.create_start(branch_name='test_lsn_mapping',
-                                            node_name='test_lsn_mapping_read',
-                                            lsn=lsn)
-        with closing(pg_here.connect()) as conn_here:
-            with conn_here.cursor() as cur_here:
-                cur_here.execute("SELECT max(x) FROM foo")
-                assert cur_here.fetchone()[0] == i
-
-        pg_here.stop_and_destroy()

test_runner/batch_others/test_recovery.py

@@ -1,64 +0,0 @@
-import os
-import time
-import psycopg2.extras
-import json
-from ast import Assert
-from contextlib import closing
-from fixtures.zenith_fixtures import ZenithEnvBuilder
-from fixtures.log_helper import log
-
-
-#
-# Test pageserver recovery after crash
-#
-def test_pageserver_recovery(zenith_env_builder: ZenithEnvBuilder):
-    zenith_env_builder.num_safekeepers = 1
-    # Override default checkpointer settings to run it more often
-    zenith_env_builder.pageserver_config_override = "tenant_config={checkpoint_distance = 1048576}"
-
-    env = zenith_env_builder.init()
-
-    # Check if failpoints enables. Otherwise the test doesn't make sense
-    f = env.zenith_cli.pageserver_enabled_features()
-
-    assert "failpoints" in f["features"], "Build pageserver with --features=failpoints option to run this test"
-    zenith_env_builder.start()
-
-    # Create a branch for us
-    env.zenith_cli.create_branch("test_pageserver_recovery", "main")
-
-    pg = env.postgres.create_start('test_pageserver_recovery')
-    log.info("postgres is running on 'test_pageserver_recovery' branch")
-
-    connstr = pg.connstr()
-
-    with closing(pg.connect()) as conn:
-        with conn.cursor() as cur:
-            with closing(env.pageserver.connect()) as psconn:
-                with psconn.cursor(cursor_factory=psycopg2.extras.DictCursor) as pscur:
-                    # Create and initialize test table
-                    cur.execute("CREATE TABLE foo(x bigint)")
-                    cur.execute("INSERT INTO foo VALUES (generate_series(1,100000))")
-
-                    # Sleep for some time to let checkpoint create image layers
-                    time.sleep(2)
-
-                    # Configure failpoints
-                    pscur.execute(
-                        "failpoints checkpoint-before-sync=sleep(2000);checkpoint-after-sync=panic")
-
-                    # Do some updates until pageserver is crashed
-                    try:
-                        while True:
-                            cur.execute("update foo set x=x+1")
-                    except Exception as err:
-                        log.info(f"Excepted server crash {err}")
-
-    log.info("Wait before server restart")
-    env.pageserver.stop()
-    env.pageserver.start()
-
-    with closing(pg.connect()) as conn:
-        with conn.cursor() as cur:
-            cur.execute("select count(*) from foo")
-            assert cur.fetchone() == (100000, )

test_runner/batch_others/test_remote_storage.py

@@ -117,7 +117,7 @@ def test_remote_storage_backup_and_restore(zenith_env_builder: ZenithEnvBuilder,
     detail = client.timeline_detail(UUID(tenant_id), UUID(timeline_id))
     assert detail['local'] is not None
     log.info("Timeline detail after attach completed: %s", detail)
-    assert lsn_from_hex(detail['local']['last_record_lsn']) >= current_lsn, 'current db Lsn should shoud not be less than the one stored on remote storage'
+    assert lsn_from_hex(detail['local']['last_record_lsn']) == current_lsn
     assert not detail['remote']['awaits_download']
 
     pg = env.postgres.create_start('main')

test_runner/fixtures/zenith_fixtures.py

@@ -980,19 +980,6 @@ class ZenithCli:
             res.check_returncode()
             return res
 
-    def pageserver_enabled_features(self) -> Any:
-        bin_pageserver = os.path.join(str(zenith_binpath), 'pageserver')
-        args = [bin_pageserver, '--enabled-features']
-        log.info('Running command "{}"'.format(' '.join(args)))
-
-        res = subprocess.run(args,
-                             check=True,
-                             universal_newlines=True,
-                             stdout=subprocess.PIPE,
-                             stderr=subprocess.PIPE)
-        log.info(f"pageserver_enabled_features success: {res.stdout}")
-        return json.loads(res.stdout)
-
     def pageserver_start(self, overrides=()) -> 'subprocess.CompletedProcess[str]':
         start_args = ['pageserver', 'start', *overrides]
         append_pageserver_param_overrides(start_args,
@@ -1303,22 +1290,6 @@ def pg_bin(test_output_dir: str) -> PgBin:
     return PgBin(test_output_dir)
 
 
-@dataclass
-class ReplayBin:
-    """A helper class for running the pageserver benchmarker tool."""
-    def run(self, tenant, timeline):
-        replay_binpath = os.path.join(str(zenith_binpath), 'replay')
-        args = [replay_binpath,
-                "--tenant", tenant.hex,
-                "--timeline", timeline.hex]
-        return subprocess.run(args)
-
-
-@pytest.fixture(scope='function')
-def replay_bin(test_output_dir):
-    return ReplayBin()
-
-
 class VanillaPostgres(PgProtocol):
     def __init__(self, pgdatadir: str, pg_bin: PgBin, port: int):
         super().__init__(host='localhost', port=port, dbname='postgres')
@@ -1411,8 +1382,8 @@ def remote_pg(test_output_dir: str) -> Iterator[RemotePostgres]:
 class ZenithProxy(PgProtocol):
     def __init__(self, port: int):
         super().__init__(host="127.0.0.1",
-                         user="proxy_user",
-                         password="pytest2",
+                         user="pytest",
+                         password="pytest",
                          port=port,
                          dbname='postgres')
         self.http_port = 7001
@@ -1428,8 +1399,8 @@ class ZenithProxy(PgProtocol):
         args = [bin_proxy]
         args.extend(["--http", f"{self.host}:{self.http_port}"])
         args.extend(["--proxy", f"{self.host}:{self.port}"])
-        args.extend(["--auth-backend", "postgres"])
-        args.extend(["--auth-endpoint", "postgres://proxy_auth:pytest1@localhost:5432/postgres"])
+        args.extend(["--auth-method", "password"])
+        args.extend(["--static-router", addr])
         self._popen = subprocess.Popen(args)
         self._wait_until_ready()
 
@@ -1451,8 +1422,7 @@ class ZenithProxy(PgProtocol):
 def static_proxy(vanilla_pg) -> Iterator[ZenithProxy]:
     """Zenith proxy that routes directly to vanilla postgres."""
     vanilla_pg.start()
-    vanilla_pg.safe_psql("create user proxy_auth with password 'pytest1' superuser")
-    vanilla_pg.safe_psql("create user proxy_user with password 'pytest2'")
+    vanilla_pg.safe_psql("create user pytest with password 'pytest';")
 
     with ZenithProxy(4432) as proxy:
         proxy.start_static()
@@ -1601,7 +1571,6 @@ class Postgres(PgProtocol):
         assert self.node_name is not None
         self.env.zenith_cli.pg_stop(self.node_name, self.tenant_id, True)
         self.node_name = None
-        self.running = False
 
         return self
 

test_runner/performance/test_pageserver.py

@@ -1,16 +0,0 @@
-from contextlib import closing
-
-import pytest
-
-from fixtures.zenith_fixtures import ZenithEnv, PgBin, ZenithEnvBuilder, DEFAULT_BRANCH_NAME, ReplayBin
-from fixtures.benchmark_fixture import MetricReport, ZenithBenchmarker
-
-
-def test_replay(zenith_env_builder: ZenithEnvBuilder,
-                  zenbenchmark: ZenithBenchmarker,
-                  replay_bin: ReplayBin):
-    env = zenith_env_builder.init_start()
-
-    tenant = env.zenith_cli.create_tenant()
-    timeline = env.zenith_cli.create_timeline("test_replay", tenant)
-    replay_bin.run(tenant, timeline)