Fix clippy errors

It should work now.

.circleci/config.yml

@@ -254,7 +254,7 @@ jobs:
           when: always
           command: |
             du -sh /tmp/test_output/*
-            find /tmp/test_output -type f ! -name "pg.log" ! -name "pageserver.log" ! -name "wal_acceptor.log" ! -name "regression.diffs" ! -name "junit.xml" ! -name "*.filediff" -delete
+            find /tmp/test_output -type f ! -name "pg.log" ! -name "pageserver.log" ! -name "wal_acceptor.log" ! -name "regression.diffs" ! -name "junit.xml" ! -name "*.filediff" ! -name "*.stdout" ! -name "*.stderr" -delete
             du -sh /tmp/test_output/*
       - store_artifacts:
           path: /tmp/test_output

.dockerignore

@@ -2,12 +2,13 @@
 **/__pycache__
 **/.pytest_cache
 
-/target
-/tmp_check
-/tmp_install
-/tmp_check_cli
-/test_output
-/.vscode
-/.zenith
-/integration_tests/.zenith
-/Dockerfile
+.git
+target
+tmp_check
+tmp_install
+tmp_check_cli
+test_output
+.vscode
+.zenith
+integration_tests/.zenith
+.mypy_cache

Cargo.lock

@@ -307,6 +307,26 @@ dependencies = [
  "vec_map",
 ]
 
+[[package]]
+name = "const_format"
+version = "0.2.21"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4556f63e28a78fa5e6f310cfea5647a25636def49a338ab69e33b34a3382057b"
+dependencies = [
+ "const_format_proc_macros",
+]
+
+[[package]]
+name = "const_format_proc_macros"
+version = "0.2.21"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "552782506c398da94466b364973b563887e0ca078bf33a76d4163736165e3594"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "unicode-xid",
+]
+
 [[package]]
 name = "control_plane"
 version = "0.1.0"
@@ -1179,10 +1199,12 @@ dependencies = [
  "bytes",
  "chrono",
  "clap",
+ "const_format",
  "crc32c",
  "daemonize",
  "futures",
  "hex",
+ "hex-literal",
  "humantime",
  "hyper",
  "lazy_static",
@@ -1406,6 +1428,7 @@ dependencies = [
  "hex",
  "md5",
  "rand",
+ "reqwest",
  "rustls",
  "serde",
  "serde_json",
@@ -2584,6 +2607,7 @@ dependencies = [
  "slog-scope",
  "slog-stdlog",
  "slog-term",
+ "tempfile",
  "thiserror",
  "tokio",
  "webpki",

Dockerfile

@@ -11,38 +11,19 @@ WORKDIR /zenith
 COPY ./vendor/postgres vendor/postgres
 COPY ./Makefile Makefile
 RUN make -j $(getconf _NPROCESSORS_ONLN) -s postgres
-
-#
-# Calculate cargo dependencies.
-# This will always run, but only generate recipe.json with list of dependencies without
-# installing them.
-#
-FROM zenithdb/build:buster AS cargo-deps-inspect
-WORKDIR /zenith
-COPY . .
-RUN cargo chef prepare --recipe-path /zenith/recipe.json
-
-#
-# Build cargo dependencies.
-# This temp cantainner should be rebuilt only if recipe.json was changed.
-#
-FROM zenithdb/build:buster AS deps-build
-WORKDIR /zenith
-COPY --from=pg-build /zenith/tmp_install/include/postgresql/server tmp_install/include/postgresql/server
-COPY --from=cargo-deps-inspect /usr/local/cargo/bin/cargo-chef /usr/local/cargo/bin/
-COPY --from=cargo-deps-inspect /zenith/recipe.json recipe.json
-RUN ROCKSDB_LIB_DIR=/usr/lib/ cargo chef cook --release --recipe-path recipe.json
+RUN rm -rf postgres_install/build
 
 #
 # Build zenith binaries
 #
+# TODO: build cargo deps as separate layer. We used cargo-chef before but that was
+# net time waste in a lot of cases. Copying Cargo.lock with empty lib.rs should do the work.
+#
 FROM zenithdb/build:buster AS build
 WORKDIR /zenith
-COPY . .
-# Copy cached dependencies
 COPY --from=pg-build /zenith/tmp_install/include/postgresql/server tmp_install/include/postgresql/server
-COPY --from=deps-build /zenith/target target
-COPY --from=deps-build /usr/local/cargo/ /usr/local/cargo/
+
+COPY . .
 RUN cargo build --release
 
 #
@@ -51,7 +32,7 @@ RUN cargo build --release
 FROM debian:buster-slim
 WORKDIR /data
 
-RUN apt-get update && apt-get -yq install librocksdb-dev libseccomp-dev openssl && \
+RUN apt-get update && apt-get -yq install libreadline-dev libseccomp-dev openssl ca-certificates && \
     mkdir zenith_install
 
 COPY --from=build /zenith/target/release/pageserver /usr/local/bin

Dockerfile.build

@@ -9,7 +9,7 @@ WORKDIR /zenith
 # Install postgres and zenith build dependencies
 # clang is for rocksdb
 RUN apt-get update && apt-get -yq install automake libtool build-essential bison flex libreadline-dev zlib1g-dev libxml2-dev \
-                                          libseccomp-dev pkg-config libssl-dev librocksdb-dev clang
+                                          libseccomp-dev pkg-config libssl-dev clang
 
 # Install rust tools
-RUN rustup component add clippy && cargo install cargo-chef cargo-audit
+RUN rustup component add clippy && cargo install cargo-audit

control_plane/src/compute.rs

@@ -452,9 +452,7 @@ impl PostgresNode {
             .output()
             .expect("failed to execute whoami");
 
-        if !output.status.success() {
-            panic!("whoami failed");
-        }
+        assert!(output.status.success(), "whoami failed");
 
         String::from_utf8(output.stdout).unwrap().trim().to_string()
     }

docs/README.md

@@ -10,5 +10,5 @@
 - [pageserver/README](/pageserver/README) — pageserver overview.
 - [postgres_ffi/README](/postgres_ffi/README) — Postgres FFI overview.
 - [test_runner/README.md](/test_runner/README.md) — tests infrastructure overview.
-- [walkeeper/README](/walkeeper/README.md) — WAL service overview.
+- [walkeeper/README](/walkeeper/README) — WAL service overview.
 - [core_changes.md](core_changes.md) - Description of Zenith changes in Postgres core

pageserver/Cargo.toml

@@ -34,8 +34,12 @@ toml = "0.5"
 scopeguard = "1.1.0"
 rust-s3 = { version = "0.27.0-rc4", features = ["no-verify-ssl"] }
 async-trait = "0.1"
+const_format = "0.2.21"
 
 postgres_ffi = { path = "../postgres_ffi" }
 zenith_metrics = { path = "../zenith_metrics" }
 zenith_utils = { path = "../zenith_utils" }
 workspace_hack = { path = "../workspace_hack" }
+
+[dev-dependencies]
+hex-literal = "0.3"

pageserver/src/bin/pageserver.rs

@@ -9,22 +9,28 @@ use std::{
     env,
     net::TcpListener,
     path::{Path, PathBuf},
-    process::exit,
     str::FromStr,
     thread,
 };
 use zenith_utils::{auth::JwtAuth, logging, postgres_backend::AuthType};
 
-use anyhow::{bail, ensure, Result};
+use anyhow::{bail, ensure, Context, Result};
 use clap::{App, Arg, ArgMatches};
 use daemonize::Daemonize;
 
 use pageserver::{
-    branches, http, page_service, tenant_mgr, PageServerConf, RelishStorageConfig, S3Config,
-    LOG_FILE_NAME,
+    branches,
+    defaults::{
+        DEFAULT_HTTP_LISTEN_ADDR, DEFAULT_PG_LISTEN_ADDR,
+        DEFAULT_RELISH_STORAGE_MAX_CONCURRENT_SYNC_LIMITS,
+    },
+    http, page_service, relish_storage, tenant_mgr, PageServerConf, RelishStorageConfig,
+    RelishStorageKind, S3Config, LOG_FILE_NAME,
 };
 use zenith_utils::http::endpoint;
 
+use const_format::formatcp;
+
 /// String arguments that can be declared via CLI or config file
 #[derive(Serialize, Deserialize)]
 struct CfgFileParams {
@@ -39,6 +45,7 @@ struct CfgFileParams {
     auth_type: Option<String>,
     // see https://github.com/alexcrichton/toml-rs/blob/6c162e6562c3e432bf04c82a3d1d789d80761a86/examples/enum_external.rs for enum deserialisation examples
     relish_storage: Option<RelishStorage>,
+    relish_storage_max_concurrent_sync: Option<String>,
 }
 
 #[derive(Serialize, Deserialize, Clone)]
@@ -89,6 +96,7 @@ impl CfgFileParams {
             auth_validation_public_key_path: get_arg("auth-validation-public-key-path"),
             auth_type: get_arg("auth-type"),
             relish_storage,
+            relish_storage_max_concurrent_sync: get_arg("relish-storage-max-concurrent-sync"),
         }
     }
 
@@ -108,6 +116,9 @@ impl CfgFileParams {
                 .or(other.auth_validation_public_key_path),
             auth_type: self.auth_type.or(other.auth_type),
             relish_storage: self.relish_storage.or(other.relish_storage),
+            relish_storage_max_concurrent_sync: self
+                .relish_storage_max_concurrent_sync
+                .or(other.relish_storage_max_concurrent_sync),
         }
     }
 
@@ -176,25 +187,34 @@ impl CfgFileParams {
             );
         }
 
-        let relish_storage_config =
-            self.relish_storage
-                .as_ref()
-                .map(|storage_params| match storage_params.clone() {
-                    RelishStorage::Local { local_path } => {
-                        RelishStorageConfig::LocalFs(PathBuf::from(local_path))
-                    }
-                    RelishStorage::AwsS3 {
-                        bucket_name,
-                        bucket_region,
-                        access_key_id,
-                        secret_access_key,
-                    } => RelishStorageConfig::AwsS3(S3Config {
-                        bucket_name,
-                        bucket_region,
-                        access_key_id,
-                        secret_access_key,
-                    }),
-                });
+        let max_concurrent_sync = match self.relish_storage_max_concurrent_sync.as_deref() {
+            Some(relish_storage_max_concurrent_sync) => {
+                relish_storage_max_concurrent_sync.parse()?
+            }
+            None => DEFAULT_RELISH_STORAGE_MAX_CONCURRENT_SYNC_LIMITS,
+        };
+        let relish_storage_config = self.relish_storage.as_ref().map(|storage_params| {
+            let storage = match storage_params.clone() {
+                RelishStorage::Local { local_path } => {
+                    RelishStorageKind::LocalFs(PathBuf::from(local_path))
+                }
+                RelishStorage::AwsS3 {
+                    bucket_name,
+                    bucket_region,
+                    access_key_id,
+                    secret_access_key,
+                } => RelishStorageKind::AwsS3(S3Config {
+                    bucket_name,
+                    bucket_region,
+                    access_key_id,
+                    secret_access_key,
+                }),
+            };
+            RelishStorageConfig {
+                max_concurrent_sync,
+                storage,
+            }
+        });
 
         Ok(PageServerConf {
             daemonize: false,
@@ -228,14 +248,14 @@ fn main() -> Result<()> {
                 .long("listen-pg")
                 .alias("listen") // keep some compatibility
                 .takes_value(true)
-                .help("listen for incoming page requests on ip:port (default: 127.0.0.1:5430)"),
+                .help(formatcp!("listen for incoming page requests on ip:port (default: {DEFAULT_PG_LISTEN_ADDR})")),
         )
         .arg(
             Arg::with_name("listen-http")
                 .long("listen-http")
                 .alias("http_endpoint") // keep some compatibility
                 .takes_value(true)
-                .help("http endpoint address for for metrics and management API calls ip:port (default: 127.0.0.1:5430)"),
+                .help(formatcp!("http endpoint address for metrics and management API calls on ip:port (default: {DEFAULT_HTTP_LISTEN_ADDR})")),
         )
         .arg(
             Arg::with_name("daemonize")
@@ -343,10 +363,19 @@ fn main() -> Result<()> {
                 .takes_value(true)
                 .help("Credentials to access the AWS S3 bucket"),
         )
+        .arg(
+            Arg::with_name("relish-storage-max-concurrent-sync")
+                .long("relish-storage-max-concurrent-sync")
+                .takes_value(true)
+                .help("Maximum allowed concurrent synchronisations with storage"),
+        )
         .get_matches();
 
     let workdir = Path::new(arg_matches.value_of("workdir").unwrap_or(".zenith"));
-    let cfg_file_path = workdir.canonicalize()?.join("pageserver.toml");
+    let cfg_file_path = workdir
+        .canonicalize()
+        .with_context(|| format!("Error opening workdir '{}'", workdir.display()))?
+        .join("pageserver.toml");
 
     let args_params = CfgFileParams::from_args(&arg_matches);
 
@@ -358,22 +387,37 @@ fn main() -> Result<()> {
         args_params
     } else {
         // Supplement the CLI arguments with the config file
-        let cfg_file_contents = std::fs::read_to_string(&cfg_file_path)?;
-        let file_params: CfgFileParams = toml::from_str(&cfg_file_contents)?;
+        let cfg_file_contents = std::fs::read_to_string(&cfg_file_path)
+            .with_context(|| format!("No pageserver config at '{}'", cfg_file_path.display()))?;
+        let file_params: CfgFileParams = toml::from_str(&cfg_file_contents).with_context(|| {
+            format!(
+                "Failed to read '{}' as pageserver config",
+                cfg_file_path.display()
+            )
+        })?;
         args_params.or(file_params)
     };
 
     // Set CWD to workdir for non-daemon modes
-    env::set_current_dir(&workdir)?;
+    env::set_current_dir(&workdir).with_context(|| {
+        format!(
+            "Failed to set application's current dir to '{}'",
+            workdir.display()
+        )
+    })?;
 
     // Ensure the config is valid, even if just init-ing
-    let mut conf = params.try_into_config()?;
+    let mut conf = params.try_into_config().with_context(|| {
+        format!(
+            "Pageserver config at '{}' is not valid",
+            cfg_file_path.display()
+        )
+    })?;
 
     conf.daemonize = arg_matches.is_present("daemonize");
 
     if init && conf.daemonize {
-        eprintln!("--daemonize cannot be used with --init");
-        exit(1);
+        bail!("--daemonize cannot be used with --init")
     }
 
     // The configuration is all set up now. Turn it into a 'static
@@ -383,16 +427,21 @@ fn main() -> Result<()> {
 
     // Create repo and exit if init was requested
     if init {
-        branches::init_pageserver(conf, create_tenant)?;
+        branches::init_pageserver(conf, create_tenant).context("Failed to init pageserver")?;
         // write the config file
-        let cfg_file_contents = toml::to_string_pretty(&params)?;
+        let cfg_file_contents = toml::to_string_pretty(&params)
+            .context("Failed to create pageserver config contents for initialisation")?;
         // TODO support enable-auth flag
-        std::fs::write(&cfg_file_path, cfg_file_contents)?;
-
-        return Ok(());
+        std::fs::write(&cfg_file_path, cfg_file_contents).with_context(|| {
+            format!(
+                "Failed to initialize pageserver config at '{}'",
+                cfg_file_path.display()
+            )
+        })?;
+        Ok(())
+    } else {
+        start_pageserver(conf).context("Failed to start pageserver")
     }
-
-    start_pageserver(conf)
 }
 
 fn start_pageserver(conf: &'static PageServerConf) -> Result<()> {
@@ -430,16 +479,20 @@ fn start_pageserver(conf: &'static PageServerConf) -> Result<()> {
 
         match daemonize.start() {
             Ok(_) => info!("Success, daemonized"),
-            Err(e) => error!("Error, {}", e),
+            Err(e) => error!("could not daemonize: {:#}", e),
         }
     }
 
+    // keep join handles for spawned threads
+    // don't spawn threads before daemonizing
+    let mut join_handles = Vec::new();
+
+    if let Some(handle) = relish_storage::run_storage_sync_thread(conf)? {
+        join_handles.push(handle);
+    }
     // Initialize tenant manager.
     tenant_mgr::init(conf);
 
-    // keep join handles for spawned threads
-    let mut join_handles = vec![];
-
     // initialize authentication for incoming connections
     let auth = match &conf.auth_type {
         AuthType::Trust | AuthType::MD5 => None,

pageserver/src/branches.rs

@@ -17,6 +17,7 @@ use std::{
 use zenith_utils::zid::{ZTenantId, ZTimelineId};
 
 use log::*;
+use zenith_utils::crashsafe_dir;
 use zenith_utils::logging;
 use zenith_utils::lsn::Lsn;
 
@@ -118,7 +119,7 @@ pub fn init_pageserver(conf: &'static PageServerConf, create_tenant: Option<&str
         println!("initializing tenantid {}", tenantid);
         create_repo(conf, tenantid, dummy_redo_mgr).with_context(|| "failed to create repo")?;
     }
-    fs::create_dir_all(conf.tenants_path())?;
+    crashsafe_dir::create_dir_all(conf.tenants_path())?;
 
     println!("pageserver init succeeded");
     Ok(())
@@ -135,12 +136,12 @@ pub fn create_repo(
     }
 
     // top-level dir may exist if we are creating it through CLI
-    fs::create_dir_all(&repo_dir)
+    crashsafe_dir::create_dir_all(&repo_dir)
         .with_context(|| format!("could not create directory {}", repo_dir.display()))?;
 
-    fs::create_dir(conf.timelines_path(&tenantid))?;
-    fs::create_dir_all(conf.branches_path(&tenantid))?;
-    fs::create_dir_all(conf.tags_path(&tenantid))?;
+    crashsafe_dir::create_dir(conf.timelines_path(&tenantid))?;
+    crashsafe_dir::create_dir_all(conf.branches_path(&tenantid))?;
+    crashsafe_dir::create_dir_all(conf.tags_path(&tenantid))?;
 
     info!("created directory structure in {}", repo_dir.display());
 
@@ -150,12 +151,13 @@ pub fn create_repo(
         conf,
         wal_redo_manager,
         tenantid,
+        false,
     ));
 
     // Load data into pageserver
     // TODO To implement zenith import we need to
     //      move data loading out of create_repo()
-    bootstrap_timeline(conf, tenantid, tli, &*repo)?;
+    bootstrap_timeline(conf, tenantid, tli, repo.as_ref())?;
 
     Ok(repo)
 }
@@ -221,11 +223,11 @@ fn bootstrap_timeline(
     // Import the contents of the data directory at the initial checkpoint
     // LSN, and any WAL after that.
     let timeline = repo.create_empty_timeline(tli)?;
-    restore_local_repo::import_timeline_from_postgres_datadir(&pgdata_path, &*timeline, lsn)?;
-
-    let wal_dir = pgdata_path.join("pg_wal");
-    restore_local_repo::import_timeline_wal(&wal_dir, &*timeline, lsn)?;
-
+    restore_local_repo::import_timeline_from_postgres_datadir(
+        &pgdata_path,
+        timeline.as_ref(),
+        lsn,
+    )?;
     timeline.checkpoint()?;
 
     println!(

pageserver/src/layered_repository.rs

@@ -22,7 +22,8 @@ use serde::{Deserialize, Serialize};
 use std::collections::hash_map::Entry;
 use std::collections::HashMap;
 use std::collections::{BTreeSet, HashSet};
-use std::fs::File;
+use std::convert::TryInto;
+use std::fs::{File, OpenOptions};
 use std::io::Write;
 use std::ops::Bound::Included;
 use std::path::{Path, PathBuf};
@@ -31,11 +32,10 @@ use std::sync::{Arc, Mutex, MutexGuard};
 use std::time::{Duration, Instant};
 use std::{fs, thread};
 
-use crate::layered_repository::inmemory_layer::FreezeLayers;
 use crate::relish::*;
-use crate::relish_storage::storage_uploader::QueueBasedRelishUploader;
+use crate::relish_storage::schedule_timeline_upload;
 use crate::repository::{GcResult, Repository, Timeline, WALRecord};
-use crate::restore_local_repo::import_timeline_wal;
+use crate::walreceiver::IS_WAL_RECEIVER;
 use crate::walredo::WalRedoManager;
 use crate::PageServerConf;
 use crate::{ZTenantId, ZTimelineId};
@@ -45,6 +45,7 @@ use zenith_metrics::{
 };
 use zenith_metrics::{register_histogram_vec, HistogramVec};
 use zenith_utils::bin_ser::BeSer;
+use zenith_utils::crashsafe_dir;
 use zenith_utils::lsn::{AtomicLsn, Lsn, RecordLsn};
 use zenith_utils::seqwait::SeqWait;
 
@@ -73,6 +74,11 @@ static ZERO_PAGE: Bytes = Bytes::from_static(&[0u8; 8192]);
 // Timeout when waiting for WAL receiver to catch up to an LSN given in a GetPage@LSN call.
 static TIMEOUT: Duration = Duration::from_secs(60);
 
+// Taken from PG_CONTROL_MAX_SAFE_SIZE
+const METADATA_MAX_SAFE_SIZE: usize = 512;
+const METADATA_CHECKSUM_SIZE: usize = std::mem::size_of::<u32>();
+const METADATA_MAX_DATA_SIZE: usize = METADATA_MAX_SAFE_SIZE - METADATA_CHECKSUM_SIZE;
+
 // Metrics collected on operations on the storage repository.
 lazy_static! {
     static ref STORAGE_TIME: HistogramVec = register_histogram_vec!(
@@ -112,7 +118,9 @@ pub struct LayeredRepository {
     timelines: Mutex<HashMap<ZTimelineId, Arc<LayeredTimeline>>>,
 
     walredo_mgr: Arc<dyn WalRedoManager + Send + Sync>,
-    relish_uploader: Option<Arc<QueueBasedRelishUploader>>,
+    /// Makes evey repo's timelines to backup their files to remote storage,
+    /// when they get frozen.
+    upload_relishes: bool,
 }
 
 /// Public interface
@@ -127,7 +135,7 @@ impl Repository for LayeredRepository {
         let mut timelines = self.timelines.lock().unwrap();
 
         // Create the timeline directory, and write initial metadata to file.
-        std::fs::create_dir_all(self.conf.timeline_path(&timelineid, &self.tenantid))?;
+        crashsafe_dir::create_dir_all(self.conf.timeline_path(&timelineid, &self.tenantid))?;
 
         let metadata = TimelineMetadata {
             disk_consistent_lsn: Lsn(0),
@@ -135,7 +143,7 @@ impl Repository for LayeredRepository {
             ancestor_timeline: None,
             ancestor_lsn: Lsn(0),
         };
-        Self::save_metadata(self.conf, timelineid, self.tenantid, &metadata)?;
+        Self::save_metadata(self.conf, timelineid, self.tenantid, &metadata, true)?;
 
         let timeline = LayeredTimeline::new(
             self.conf,
@@ -144,8 +152,8 @@ impl Repository for LayeredRepository {
             timelineid,
             self.tenantid,
             Arc::clone(&self.walredo_mgr),
-            self.relish_uploader.as_ref().map(Arc::clone),
             0,
+            false,
         )?;
 
         let timeline_rc = Arc::new(timeline);
@@ -179,8 +187,8 @@ impl Repository for LayeredRepository {
             ancestor_timeline: Some(src),
             ancestor_lsn: start_lsn,
         };
-        std::fs::create_dir_all(self.conf.timeline_path(&dst, &self.tenantid))?;
-        Self::save_metadata(self.conf, dst, self.tenantid, &metadata)?;
+        crashsafe_dir::create_dir_all(self.conf.timeline_path(&dst, &self.tenantid))?;
+        Self::save_metadata(self.conf, dst, self.tenantid, &metadata, true)?;
 
         info!("branched timeline {} from {} at {}", dst, src, start_lsn);
 
@@ -217,6 +225,7 @@ impl LayeredRepository {
             Some(timeline) => Ok(timeline.clone()),
             None => {
                 let metadata = Self::load_metadata(self.conf, timelineid, self.tenantid)?;
+                let disk_consistent_lsn = metadata.disk_consistent_lsn;
 
                 // Recurse to look up the ancestor timeline.
                 //
@@ -236,12 +245,12 @@ impl LayeredRepository {
                     timelineid,
                     self.tenantid,
                     Arc::clone(&self.walredo_mgr),
-                    self.relish_uploader.as_ref().map(Arc::clone),
-                    0, // init with 0 and update after layers are loaded
+                    0, // init with 0 and update after layers are loaded,
+                    self.upload_relishes,
                 )?;
 
                 // List the layers on disk, and load them into the layer map
-                timeline.load_layer_map()?;
+                timeline.load_layer_map(disk_consistent_lsn)?;
 
                 // needs to be after load_layer_map
                 timeline.init_current_logical_size()?;
@@ -254,11 +263,6 @@ impl LayeredRepository {
                     timelineid,
                     timeline.get_last_record_lsn()
                 );
-                let wal_dir = self
-                    .conf
-                    .timeline_path(&timelineid, &self.tenantid)
-                    .join("wal");
-                import_timeline_wal(&wal_dir, timeline.as_ref(), timeline.get_last_record_lsn())?;
 
                 if cfg!(debug_assertions) {
                     // check again after wal loading
@@ -275,15 +279,14 @@ impl LayeredRepository {
         conf: &'static PageServerConf,
         walredo_mgr: Arc<dyn WalRedoManager + Send + Sync>,
         tenantid: ZTenantId,
+        upload_relishes: bool,
     ) -> LayeredRepository {
         LayeredRepository {
             tenantid,
             conf,
             timelines: Mutex::new(HashMap::new()),
             walredo_mgr,
-            relish_uploader: conf.relish_storage_config.as_ref().map(|config| {
-                Arc::new(QueueBasedRelishUploader::new(config, &conf.workdir).unwrap())
-            }),
+            upload_relishes,
         }
     }
 
@@ -358,13 +361,36 @@ impl LayeredRepository {
         timelineid: ZTimelineId,
         tenantid: ZTenantId,
         data: &TimelineMetadata,
+        first_save: bool,
     ) -> Result<PathBuf> {
-        let path = conf.timeline_path(&timelineid, &tenantid).join("metadata");
-        let mut file = File::create(&path)?;
+        let timeline_path = conf.timeline_path(&timelineid, &tenantid);
+        let path = timeline_path.join("metadata");
+        // use OpenOptions to ensure file presence is consistent with first_save
+        let mut file = OpenOptions::new()
+            .write(true)
+            .create_new(first_save)
+            .open(&path)?;
 
         info!("saving metadata {}", path.display());
 
-        file.write_all(&TimelineMetadata::ser(data)?)?;
+        let mut metadata_bytes = TimelineMetadata::ser(data)?;
+
+        assert!(metadata_bytes.len() <= METADATA_MAX_DATA_SIZE);
+        metadata_bytes.resize(METADATA_MAX_SAFE_SIZE, 0u8);
+
+        let checksum = crc32c::crc32c(&metadata_bytes[..METADATA_MAX_DATA_SIZE]);
+        metadata_bytes[METADATA_MAX_DATA_SIZE..].copy_from_slice(&u32::to_le_bytes(checksum));
+
+        if file.write(&metadata_bytes)? != metadata_bytes.len() {
+            bail!("Could not write all the metadata bytes in a single call");
+        }
+        file.sync_all()?;
+
+        // fsync the parent directory to ensure the directory entry is durable
+        if first_save {
+            let timeline_dir = File::open(&timeline_path)?;
+            timeline_dir.sync_all()?;
+        }
 
         Ok(path)
     }
@@ -375,9 +401,18 @@ impl LayeredRepository {
         tenantid: ZTenantId,
     ) -> Result<TimelineMetadata> {
         let path = conf.timeline_path(&timelineid, &tenantid).join("metadata");
-        let data = std::fs::read(&path)?;
+        let metadata_bytes = std::fs::read(&path)?;
+        ensure!(metadata_bytes.len() == METADATA_MAX_SAFE_SIZE);
 
-        let data = TimelineMetadata::des(&data)?;
+        let data = &metadata_bytes[..METADATA_MAX_DATA_SIZE];
+        let calculated_checksum = crc32c::crc32c(data);
+
+        let checksum_bytes: &[u8; METADATA_CHECKSUM_SIZE] =
+            metadata_bytes[METADATA_MAX_DATA_SIZE..].try_into()?;
+        let expected_checksum = u32::from_le_bytes(*checksum_bytes);
+        ensure!(calculated_checksum == expected_checksum);
+
+        let data = TimelineMetadata::des_prefix(data)?;
         assert!(data.disk_consistent_lsn.is_aligned());
 
         Ok(data)
@@ -507,7 +542,7 @@ impl LayeredRepository {
                     error!("timeline size calculation diverged, incremental doesn't match non incremental. incremental={} non_incremental={}", incremental, non_incremental);
                 }
             }
-            Err(e) => error!("failed to calculate non incremental timeline size: {}", e),
+            Err(e) => error!("failed to calculate non incremental timeline size: {:#}", e),
         }
     }
 
@@ -553,8 +588,6 @@ pub struct LayeredTimeline {
     // WAL redo manager
     walredo_mgr: Arc<dyn WalRedoManager + Sync + Send>,
 
-    relish_uploader: Option<Arc<QueueBasedRelishUploader>>,
-
     // What page versions do we hold in the repository? If we get a
     // request > last_record_lsn, we need to wait until we receive all
     // the WAL up to the request. The SeqWait provides functions for
@@ -602,6 +635,9 @@ pub struct LayeredTimeline {
     // TODO: it is possible to combine these two fields into single one using custom metric which uses SeqCst
     // ordering for its operations, but involves private modules, and macro trickery
     current_logical_size_gauge: IntGauge,
+
+    /// If `true`, will backup its timeline files to remote storage after freezing.
+    upload_relishes: bool,
 }
 
 /// Public interface functions
@@ -609,8 +645,11 @@ impl Timeline for LayeredTimeline {
     /// Wait until WAL has been received up to the given LSN.
     fn wait_lsn(&self, lsn: Lsn) -> Result<()> {
         // This should never be called from the WAL receiver thread, because that could lead
-        // to a deadlock. FIXME: Is there a less hacky way to check that?
-        assert_ne!(thread::current().name(), Some("WAL receiver thread"));
+        // to a deadlock.
+        assert!(
+            !IS_WAL_RECEIVER.with(|c| c.get()),
+            "wait_lsn called by WAL receiver thread"
+        );
 
         self.last_record_lsn
             .wait_for_timeout(lsn, TIMEOUT)
@@ -982,8 +1021,8 @@ impl LayeredTimeline {
         timelineid: ZTimelineId,
         tenantid: ZTenantId,
         walredo_mgr: Arc<dyn WalRedoManager + Send + Sync>,
-        relish_uploader: Option<Arc<QueueBasedRelishUploader>>,
         current_logical_size: usize,
+        upload_relishes: bool,
     ) -> Result<LayeredTimeline> {
         let current_logical_size_gauge = LOGICAL_TIMELINE_SIZE
             .get_metric_with_label_values(&[&tenantid.to_string(), &timelineid.to_string()])
@@ -995,7 +1034,6 @@ impl LayeredTimeline {
             layers: Mutex::new(LayerMap::default()),
 
             walredo_mgr,
-            relish_uploader,
 
             // initialize in-memory 'last_record_lsn' from 'disk_consistent_lsn'.
             last_record_lsn: SeqWait::new(RecordLsn {
@@ -1008,6 +1046,7 @@ impl LayeredTimeline {
             ancestor_lsn: metadata.ancestor_lsn,
             current_logical_size: AtomicUsize::new(current_logical_size),
             current_logical_size_gauge,
+            upload_relishes,
         };
         Ok(timeline)
     }
@@ -1015,7 +1054,7 @@ impl LayeredTimeline {
     ///
     /// Scan the timeline directory to populate the layer map
     ///
-    fn load_layer_map(&self) -> anyhow::Result<()> {
+    fn load_layer_map(&self, disk_consistent_lsn: Lsn) -> anyhow::Result<()> {
         info!(
             "loading layer map for timeline {} into memory",
             self.timelineid
@@ -1024,8 +1063,20 @@ impl LayeredTimeline {
         let (imgfilenames, mut deltafilenames) =
             filename::list_files(self.conf, self.timelineid, self.tenantid)?;
 
+        let timeline_path = self.conf.timeline_path(&self.timelineid, &self.tenantid);
+
         // First create ImageLayer structs for each image file.
         for filename in imgfilenames.iter() {
+            if filename.lsn > disk_consistent_lsn {
+                warn!(
+                    "found future image layer {} on timeline {}",
+                    filename, self.timelineid
+                );
+
+                rename_to_backup(timeline_path.join(filename.to_string()))?;
+                continue;
+            }
+
             let layer = ImageLayer::new(self.conf, self.timelineid, self.tenantid, filename);
 
             info!(
@@ -1043,6 +1094,17 @@ impl LayeredTimeline {
         deltafilenames.sort();
 
         for filename in deltafilenames.iter() {
+            ensure!(filename.start_lsn < filename.end_lsn);
+            if filename.end_lsn > disk_consistent_lsn {
+                warn!(
+                    "found future delta layer {} on timeline {}",
+                    filename, self.timelineid
+                );
+
+                rename_to_backup(timeline_path.join(filename.to_string()))?;
+                continue;
+            }
+
             let predecessor = layers.get(&filename.seg, filename.start_lsn);
 
             let predecessor_str: String = if let Some(prec) = &predecessor {
@@ -1190,13 +1252,12 @@ impl LayeredTimeline {
         assert!(lsn.is_aligned());
 
         let last_record_lsn = self.get_last_record_lsn();
-        if lsn <= last_record_lsn {
-            panic!(
-                "cannot modify relation after advancing last_record_lsn (incoming_lsn={}, last_record_lsn={})",
-                lsn,
-                last_record_lsn
-            );
-        }
+        assert!(
+            lsn > last_record_lsn,
+            "cannot modify relation after advancing last_record_lsn (incoming_lsn={}, last_record_lsn={})",
+            lsn,
+            last_record_lsn,
+        );
 
         // Do we have a layer open for writing already?
         let layer;
@@ -1316,6 +1377,8 @@ impl LayeredTimeline {
             last_record_lsn
         );
 
+        let timeline_dir = File::open(self.conf.timeline_path(&self.timelineid, &self.tenantid))?;
+
         // Take the in-memory layer with the oldest WAL record. If it's older
         // than the threshold, write it out to disk as a new image and delta file.
         // Repeat until all remaining in-memory layers are within the threshold.
@@ -1327,6 +1390,8 @@ impl LayeredTimeline {
         // a lot of memory and/or aren't receiving much updates anymore.
         let mut disk_consistent_lsn = last_record_lsn;
 
+        let mut created_historics = false;
+
         while let Some((oldest_layer, oldest_generation)) = layers.peek_oldest_open() {
             let oldest_pending_lsn = oldest_layer.get_oldest_pending_lsn();
 
@@ -1352,59 +1417,14 @@ impl LayeredTimeline {
                 break;
             }
 
-            // Freeze the layer.
-            //
-            // This is a two-step process. First, we "freeze" the in-memory
-            // layer, to close it for new writes, and replace the original
-            // layer with the new frozen in-memory layer (and possibly a new
-            // open layer to hold changes newer than the cutoff.) Then we write
-            // the frozen layer to disk, and replace the in-memory frozen layer
-            // with the new on-disk layers.
-            let FreezeLayers {
-                frozen,
-                open: maybe_new_open,
-            } = oldest_layer.freeze(last_record_lsn)?;
-
             // replace this layer with the new layers that 'freeze' returned
             layers.pop_oldest_open();
-            if let Some(new_open) = maybe_new_open.clone() {
-                layers.insert_open(new_open);
-            }
 
-            // We temporarily insert InMemory layer into historic list here.
-            // TODO: check that all possible concurrent users of 'historic' treat it right
-            layers.insert_historic(frozen.clone());
+            let new_delta_layer = oldest_layer.write_to_disk(self)?;
+            created_historics = true;
 
-            // Write the now-frozen layer to disk. That could take a while, so release the lock while do it
-            drop(layers);
-            let new_historics = frozen.write_to_disk(self)?;
-            layers = self.layers.lock().unwrap();
-            if let Some(relish_uploader) = &self.relish_uploader {
-                for label_path in new_historics.iter().filter_map(|layer| layer.path()) {
-                    relish_uploader.schedule_upload(self.timelineid, label_path);
-                }
-            }
-
-            // Finally, replace the frozen in-memory layer with the new on-disk layers
-            layers.remove_historic(frozen.clone());
-
-            // If we created a successor InMemoryLayer, its predecessor is
-            // currently the frozen layer. We need to update the predecessor
-            // to be the latest on-disk layer.
-            if let Some(last_historic) = new_historics.last() {
-                if let Some(new_open) = &maybe_new_open {
-                    let maybe_old_predecessor =
-                        new_open.update_predecessor(Arc::clone(last_historic));
-                    let old_predecessor = maybe_old_predecessor
-                        .expect("new_open should always be a successor to frozen");
-                    assert!(layer_ptr_eq(frozen.as_ref(), old_predecessor.as_ref()));
-                }
-            }
-
-            // Add the historics to the LayerMap
-            for n in new_historics {
-                layers.insert_historic(n);
-            }
+            // Add the historic to the LayerMap
+            layers.insert_historic(new_delta_layer);
         }
 
         // Call unload() on all frozen layers, to release memory.
@@ -1414,6 +1434,14 @@ impl LayeredTimeline {
             layer.unload()?;
         }
 
+        drop(layers);
+
+        if created_historics {
+            // We must fsync the timeline dir to ensure the directory entries for
+            // new layer files are durable
+            timeline_dir.sync_all()?;
+        }
+
         // Save the metadata, with updated 'disk_consistent_lsn', to a
         // file in the timeline dir. After crash, we will restart WAL
         // streaming and processing from that point.
@@ -1438,10 +1466,23 @@ impl LayeredTimeline {
             ancestor_timeline: ancestor_timelineid,
             ancestor_lsn: self.ancestor_lsn,
         };
-        let metadata_path =
-            LayeredRepository::save_metadata(self.conf, self.timelineid, self.tenantid, &metadata)?;
-        if let Some(relish_uploader) = &self.relish_uploader {
-            relish_uploader.schedule_upload(self.timelineid, metadata_path);
+        let _metadata_path = LayeredRepository::save_metadata(
+            self.conf,
+            self.timelineid,
+            self.tenantid,
+            &metadata,
+            false,
+        )?;
+        if self.upload_relishes {
+            schedule_timeline_upload(())
+            // schedule_timeline_upload(LocalTimeline {
+            //     tenant_id: self.tenantid,
+            //     timeline_id: self.timelineid,
+            //     metadata_path,
+            //     image_layers: image_layer_uploads,
+            //     delta_layers: delta_layer_uploads,
+            //     disk_consistent_lsn,
+            // });
         }
 
         // Also update the in-memory copy
@@ -1481,181 +1522,212 @@ impl LayeredTimeline {
             self.timelineid, cutoff
         );
         info!("retain_lsns:  {:?}", retain_lsns);
-
-        let mut layers_to_remove: Vec<Arc<dyn Layer>> = Vec::new();
-
-        // Scan all on-disk layers in the timeline.
-        //
-        // Garbage collect the layer if all conditions are satisfied:
-        // 1. it is older than cutoff LSN;
-        // 2. it doesn't need to be retained for 'retain_lsns';
-        // 3. newer on-disk layer exists (only for non-dropped segments);
-        // 4. this layer doesn't serve as a tombstone for some older layer;
-        //
         let mut layers = self.layers.lock().unwrap();
-        'outer: for l in layers.iter_historic_layers() {
-            let seg = l.get_seg_tag();
 
-            if seg.rel.is_relation() {
-                result.ondisk_relfiles_total += 1;
-            } else {
-                result.ondisk_nonrelfiles_total += 1;
-            }
+        'outer: loop {
+            let mut layers_to_remove: Vec<Arc<dyn Layer>> = Vec::new();
+            let mut layer_to_materialize: Option<Arc<dyn Layer>> = None;
+
+            // Scan all on-disk layers in the timeline.
+            //
+            // Garbage collect the layer if all conditions are satisfied:
+            // 1. it is older than cutoff LSN;
+            // 2. it doesn't need to be retained for 'retain_lsns';
+            // 3. newer on-disk layer exists (only for non-dropped segments);
+            // 4. this layer doesn't serve as a tombstone for some older layer;
+            //
+            'for_all_historic_layers: for l in layers.iter_historic_layers() {
+                let seg = l.get_seg_tag();
 
-            // 1. Is it newer than cutoff point?
-            if l.get_end_lsn() > cutoff {
-                info!(
-                    "keeping {} {}-{} because it's newer than cutoff {}",
-                    seg,
-                    l.get_start_lsn(),
-                    l.get_end_lsn(),
-                    cutoff
-                );
                 if seg.rel.is_relation() {
-                    result.ondisk_relfiles_needed_by_cutoff += 1;
+                    result.ondisk_relfiles_total += 1;
                 } else {
-                    result.ondisk_nonrelfiles_needed_by_cutoff += 1;
+                    result.ondisk_nonrelfiles_total += 1;
                 }
-                continue 'outer;
-            }
 
-            // 2. Is it needed by a child branch?
-            for retain_lsn in &retain_lsns {
-                // start_lsn is inclusive and end_lsn is exclusive
-                if l.get_start_lsn() <= *retain_lsn && *retain_lsn < l.get_end_lsn() {
+                // 1. Create image layer for this relation?
+                if !l.is_dropped()
+                    && !layers.newer_image_layer_exists(l.get_seg_tag(), l.get_end_lsn())
+                {
+                    if l.is_incremental() {
+                        layer_to_materialize = Some(Arc::clone(&l));
+                        break 'for_all_historic_layers;
+                    } else {
+                        info!(
+                            "keeping {} {}-{} because it's newer than cutoff {}",
+                            seg,
+                            l.get_start_lsn(),
+                            l.get_end_lsn(),
+                            cutoff
+                        );
+                        if seg.rel.is_relation() {
+                            result.ondisk_relfiles_needed_by_cutoff += 1;
+                        } else {
+                            result.ondisk_nonrelfiles_needed_by_cutoff += 1;
+                        }
+                        continue 'for_all_historic_layers;
+                    }
+                }
+
+                // 2. Is it newer than cutoff point?
+                if l.get_end_lsn() > cutoff {
                     info!(
-                        "keeping {} {}-{} because it's needed by branch point {}",
+                        "keeping {} {}-{} because it's newer than cutoff {}",
                         seg,
                         l.get_start_lsn(),
                         l.get_end_lsn(),
-                        *retain_lsn
+                        cutoff
                     );
                     if seg.rel.is_relation() {
-                        result.ondisk_relfiles_needed_by_branches += 1;
+                        result.ondisk_relfiles_needed_by_cutoff += 1;
                     } else {
-                        result.ondisk_nonrelfiles_needed_by_branches += 1;
+                        result.ondisk_nonrelfiles_needed_by_cutoff += 1;
                     }
-                    continue 'outer;
+                    continue 'for_all_historic_layers;
                 }
-            }
 
-            // 3. Is there a later on-disk layer for this relation?
-            if !l.is_dropped() && !layers.newer_image_layer_exists(l.get_seg_tag(), l.get_end_lsn())
-            {
-                info!(
-                    "keeping {} {}-{} because it is the latest layer",
-                    seg,
-                    l.get_start_lsn(),
-                    l.get_end_lsn()
-                );
-                if seg.rel.is_relation() {
-                    result.ondisk_relfiles_not_updated += 1;
-                } else {
-                    result.ondisk_nonrelfiles_not_updated += 1;
-                }
-                continue 'outer;
-            }
-
-            // 4. Does this layer serve as a tombstome for some older layer?
-            if l.is_dropped() {
-                let prior_lsn = l.get_start_lsn().checked_sub(1u64).unwrap();
-
-                // Check if this layer serves as a tombstone for this timeline
-                // We have to do this separately from timeline check below,
-                // because LayerMap of this timeline is already locked.
-                let mut is_tombstone = layers.layer_exists_at_lsn(l.get_seg_tag(), prior_lsn)?;
-                if is_tombstone {
-                    info!(
-                        "earlier layer exists at {} in {}",
-                        prior_lsn, self.timelineid
-                    );
-                }
-                // Now check ancestor timelines, if any
-                else if let Some(ancestor) = &self.ancestor_timeline {
-                    let prior_lsn = ancestor.get_last_record_lsn();
-                    if seg.rel.is_blocky() {
+                // 3. Is it needed by a child branch?
+                for retain_lsn in &retain_lsns {
+                    // start_lsn is inclusive and end_lsn is exclusive
+                    if l.get_start_lsn() <= *retain_lsn && *retain_lsn < l.get_end_lsn() {
                         info!(
-                            "check blocky relish size {} at {} in {} for layer {}-{}",
+                            "keeping {} {}-{} because it's needed by branch point {}",
                             seg,
-                            prior_lsn,
-                            ancestor.timelineid,
                             l.get_start_lsn(),
-                            l.get_end_lsn()
+                            l.get_end_lsn(),
+                            *retain_lsn
                         );
-                        match ancestor.get_relish_size(seg.rel, prior_lsn).unwrap() {
-                            Some(size) => {
-                                let last_live_seg = SegmentTag::from_blknum(seg.rel, size - 1);
-                                info!(
-                                    "blocky rel size is {} last_live_seg.segno {} seg.segno {}",
-                                    size, last_live_seg.segno, seg.segno
-                                );
-                                if last_live_seg.segno >= seg.segno {
-                                    is_tombstone = true;
+                        if seg.rel.is_relation() {
+                            result.ondisk_relfiles_needed_by_branches += 1;
+                        } else {
+                            result.ondisk_nonrelfiles_needed_by_branches += 1;
+                        }
+                        continue 'for_all_historic_layers;
+                    }
+                }
+
+                // 4. Does this layer serve as a tombstome for some older layer?
+                if l.is_dropped() {
+                    let prior_lsn = l.get_start_lsn().checked_sub(1u64).unwrap();
+
+                    // Check if this layer serves as a tombstone for this timeline
+                    // We have to do this separately from timeline check below,
+                    // because LayerMap of this timeline is already locked.
+                    let mut is_tombstone =
+                        layers.layer_exists_at_lsn(l.get_seg_tag(), prior_lsn)?;
+                    if is_tombstone {
+                        info!(
+                            "earlier layer exists at {} in {}",
+                            prior_lsn, self.timelineid
+                        );
+                    }
+                    // Now check ancestor timelines, if any
+                    else if let Some(ancestor) = &self.ancestor_timeline {
+                        let prior_lsn = ancestor.get_last_record_lsn();
+                        if seg.rel.is_blocky() {
+                            info!(
+                                "check blocky relish size {} at {} in {} for layer {}-{}",
+                                seg,
+                                prior_lsn,
+                                ancestor.timelineid,
+                                l.get_start_lsn(),
+                                l.get_end_lsn()
+                            );
+                            match ancestor.get_relish_size(seg.rel, prior_lsn).unwrap() {
+                                Some(size) => {
+                                    let last_live_seg = SegmentTag::from_blknum(seg.rel, size - 1);
+                                    info!(
+                                        "blocky rel size is {} last_live_seg.segno {} seg.segno {}",
+                                        size, last_live_seg.segno, seg.segno
+                                    );
+                                    if last_live_seg.segno >= seg.segno {
+                                        is_tombstone = true;
+                                    }
+                                }
+                                _ => {
+                                    info!("blocky rel doesn't exist");
                                 }
                             }
-                            _ => {
-                                info!("blocky rel doesn't exist");
-                            }
+                        } else {
+                            info!(
+                                "check non-blocky relish existence {} at {} in {} for layer {}-{}",
+                                seg,
+                                prior_lsn,
+                                ancestor.timelineid,
+                                l.get_start_lsn(),
+                                l.get_end_lsn()
+                            );
+                            is_tombstone =
+                                ancestor.get_rel_exists(seg.rel, prior_lsn).unwrap_or(false);
                         }
-                    } else {
+                    }
+
+                    if is_tombstone {
                         info!(
-                            "check non-blocky relish existence {} at {} in {} for layer {}-{}",
-                            seg,
-                            prior_lsn,
-                            ancestor.timelineid,
-                            l.get_start_lsn(),
-                            l.get_end_lsn()
-                        );
-                        is_tombstone = ancestor.get_rel_exists(seg.rel, prior_lsn).unwrap_or(false);
+							"keeping {} {}-{} because this layer servers as a tombstome for older layer",
+							seg,
+							l.get_start_lsn(),
+							l.get_end_lsn()
+						);
+
+                        if seg.rel.is_relation() {
+                            result.ondisk_relfiles_needed_as_tombstone += 1;
+                        } else {
+                            result.ondisk_nonrelfiles_needed_as_tombstone += 1;
+                        }
+                        continue 'for_all_historic_layers;
                     }
                 }
 
-                if is_tombstone {
-                    info!(
-                        "keeping {} {}-{} because this layer servers as a tombstome for older layer",
-                        seg,
-                        l.get_start_lsn(),
-                        l.get_end_lsn()
-                    );
+                // We didn't find any reason to keep this file, so remove it.
+                info!(
+                    "garbage collecting {} {}-{} {}",
+                    l.get_seg_tag(),
+                    l.get_start_lsn(),
+                    l.get_end_lsn(),
+                    l.is_dropped()
+                );
+                layers_to_remove.push(Arc::clone(&l));
+            }
 
-                    if seg.rel.is_relation() {
-                        result.ondisk_relfiles_needed_as_tombstone += 1;
-                    } else {
-                        result.ondisk_nonrelfiles_needed_as_tombstone += 1;
-                    }
-                    continue 'outer;
+            // Actually delete the layers from disk and remove them from the map.
+            // (couldn't do this in the loop above, because you cannot modify a collection
+            // while iterating it. BTreeMap::retain() would be another option)
+            for doomed_layer in layers_to_remove {
+                doomed_layer.delete()?;
+                layers.remove_historic(doomed_layer.clone());
+
+                match (
+                    doomed_layer.is_dropped(),
+                    doomed_layer.get_seg_tag().rel.is_relation(),
+                ) {
+                    (true, true) => result.ondisk_relfiles_dropped += 1,
+                    (true, false) => result.ondisk_nonrelfiles_dropped += 1,
+                    (false, true) => result.ondisk_relfiles_removed += 1,
+                    (false, false) => result.ondisk_nonrelfiles_removed += 1,
                 }
             }
 
-            // We didn't find any reason to keep this file, so remove it.
-            info!(
-                "garbage collecting {} {}-{} {}",
-                l.get_seg_tag(),
-                l.get_start_lsn(),
-                l.get_end_lsn(),
-                l.is_dropped()
-            );
-            layers_to_remove.push(Arc::clone(&l));
-        }
-
-        // Actually delete the layers from disk and remove them from the map.
-        // (couldn't do this in the loop above, because you cannot modify a collection
-        // while iterating it. BTreeMap::retain() would be another option)
-        for doomed_layer in layers_to_remove {
-            doomed_layer.delete()?;
-            layers.remove_historic(doomed_layer.clone());
-
-            match (
-                doomed_layer.is_dropped(),
-                doomed_layer.get_seg_tag().rel.is_relation(),
-            ) {
-                (true, true) => result.ondisk_relfiles_dropped += 1,
-                (true, false) => result.ondisk_nonrelfiles_dropped += 1,
-                (false, true) => result.ondisk_relfiles_removed += 1,
-                (false, false) => result.ondisk_nonrelfiles_removed += 1,
+            if let Some(delta_layer) = layer_to_materialize {
+                drop(layers); // release lock, as far as new image layers are created only by GC thread,
+                let image_layer = ImageLayer::create_from_src(
+                    self.conf,
+                    &self,
+                    &*delta_layer,
+                    delta_layer.get_end_lsn(),
+                )?;
+                layers = self.layers.lock().unwrap();
+                info!(
+                    "materialize layer {} {}-{}",
+                    delta_layer.get_seg_tag(),
+                    delta_layer.get_start_lsn(),
+                    delta_layer.get_end_lsn()
+                );
+                layers.insert_historic(Arc::new(image_layer));
+                continue 'outer;
             }
+            break 'outer;
         }
-
         result.elapsed = now.elapsed();
         Ok(result)
     }
@@ -1864,13 +1936,22 @@ pub fn dump_layerfile_from_path(path: &Path) -> Result<()> {
     Ok(())
 }
 
-/// Check for equality of Layer memory addresses
-fn layer_ptr_eq(l1: &dyn Layer, l2: &dyn Layer) -> bool {
-    let l1_ptr = l1 as *const dyn Layer;
-    let l2_ptr = l2 as *const dyn Layer;
-    // comparing *const dyn Layer will not only check for data address equality,
-    // but also for vtable address equality.
-    // to avoid this, we compare *const ().
-    // see here for more https://github.com/rust-lang/rust/issues/46139
-    std::ptr::eq(l1_ptr as *const (), l2_ptr as *const ())
+/// Add a suffix to a layer file's name: .{num}.old
+/// Uses the first available num (starts at 0)
+fn rename_to_backup(path: PathBuf) -> anyhow::Result<()> {
+    let filename = path.file_name().unwrap().to_str().unwrap();
+    let mut new_path = path.clone();
+
+    for i in 0u32.. {
+        new_path.set_file_name(format!("{}.{}.old", filename, i));
+        if !new_path.exists() {
+            std::fs::rename(&path, &new_path)?;
+            return Ok(());
+        }
+    }
+
+    Err(anyhow!(
+        "couldn't find an unused backup number for {:?}",
+        path
+    ))
 }

pageserver/src/layered_repository/delta_layer.rs

@@ -381,14 +381,15 @@ impl DeltaLayer {
         }
     }
 
-    /// Create a new delta file, using the given btreemaps containing the page versions and
-    /// relsizes.
+    /// Create a new delta file, using the given page versions and relsizes.
+    /// The page versions are passed by an iterator; the iterator must return
+    /// page versions in blknum+lsn order.
     ///
     /// This is used to write the in-memory layer to disk. The in-memory layer uses the same
     /// data structure with two btreemaps as we do, so passing the btreemaps is currently
     /// expedient.
     #[allow(clippy::too_many_arguments)]
-    pub fn create(
+    pub fn create<'a>(
         conf: &'static PageServerConf,
         timelineid: ZTimelineId,
         tenantid: ZTenantId,
@@ -397,7 +398,7 @@ impl DeltaLayer {
         end_lsn: Lsn,
         dropped: bool,
         predecessor: Option<Arc<dyn Layer>>,
-        page_versions: BTreeMap<(u32, Lsn), PageVersion>,
+        page_versions: impl Iterator<Item = (&'a (u32, Lsn), &'a PageVersion)>,
         relsizes: BTreeMap<Lsn, u32>,
     ) -> Result<DeltaLayer> {
         let delta_layer = DeltaLayer {
@@ -433,19 +434,21 @@ impl DeltaLayer {
         for (key, page_version) in page_versions {
             let page_image_range = page_version
                 .page_image
-                .map(|page_image| page_version_writer.write_blob(page_image.as_ref()))
+                .as_ref()
+                .map(|page_image| page_version_writer.write_blob(page_image))
                 .transpose()?;
 
             let record_range = page_version
                 .record
+                .as_ref()
                 .map(|record| {
-                    let buf = WALRecord::ser(&record)?;
+                    let buf = WALRecord::ser(record)?;
                     page_version_writer.write_blob(&buf)
                 })
                 .transpose()?;
 
             let old = inner.page_version_metas.insert(
-                key,
+                *key,
                 PageVersionMeta {
                     page_image_range,
                     record_range,
@@ -484,7 +487,8 @@ impl DeltaLayer {
         let book = chapter.close()?;
 
         // This flushes the underlying 'buf_writer'.
-        book.close()?;
+        let writer = book.close()?;
+        writer.get_ref().sync_all()?;
 
         trace!("saved {}", &path.display());
 

pageserver/src/layered_repository/filename.rs

@@ -290,7 +290,11 @@ pub fn list_files(
             deltafiles.push(deltafilename);
         } else if let Some(imgfilename) = ImageFileName::from_str(fname) {
             imgfiles.push(imgfilename);
-        } else if fname == "wal" || fname == "metadata" || fname == "ancestor" {
+        } else if fname == "wal"
+            || fname == "metadata"
+            || fname == "ancestor"
+            || fname.ends_with(".old")
+        {
             // ignore these
         } else {
             warn!("unrecognized filename in timeline dir: {}", fname);

pageserver/src/layered_repository/image_layer.rs

@@ -337,7 +337,8 @@ impl ImageLayer {
         let book = chapter.close()?;
 
         // This flushes the underlying 'buf_writer'.
-        book.close()?;
+        let writer = book.close()?;
+        writer.get_ref().sync_all()?;
 
         trace!("saved {}", &path.display());
 

pageserver/src/layered_repository/inmemory_layer.rs

@@ -15,13 +15,11 @@ use crate::{ZTenantId, ZTimelineId};
 use anyhow::{bail, Result};
 use bytes::Bytes;
 use log::*;
-use std::cmp::min;
 use std::collections::BTreeMap;
 use std::ops::Bound::Included;
 use std::path::PathBuf;
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, RwLock};
 
-use zenith_utils::accum::Accum;
 use zenith_utils::lsn::Lsn;
 
 pub struct InMemoryLayer {
@@ -36,21 +34,21 @@ pub struct InMemoryLayer {
     ///
     start_lsn: Lsn,
 
-    /// Frozen in-memory layers have an inclusive end LSN.
-    end_lsn: Option<Lsn>,
-
     /// LSN of the oldest page version stored in this layer
     oldest_pending_lsn: Lsn,
 
     /// The above fields never change. The parts that do change are in 'inner',
     /// and protected by mutex.
-    inner: Mutex<InMemoryLayerInner>,
+    inner: RwLock<InMemoryLayerInner>,
 }
 
 pub struct InMemoryLayerInner {
     /// If this relation was dropped, remember when that happened.
     drop_lsn: Option<Lsn>,
 
+    /// LSN of last record
+    end_lsn: Option<Lsn>,
+
     ///
     /// All versions of all pages in the layer are are kept here.
     /// Indexed by block number and LSN.
@@ -95,7 +93,7 @@ impl Layer for InMemoryLayer {
     // An in-memory layer doesn't really have a filename as it's not stored on disk,
     // but we construct a filename as if it was a delta layer
     fn filename(&self) -> PathBuf {
-        let inner = self.inner.lock().unwrap();
+        let inner = self.inner.read().unwrap();
 
         let end_lsn;
         let dropped;
@@ -118,6 +116,11 @@ impl Layer for InMemoryLayer {
         PathBuf::from(format!("inmem-{}", delta_filename))
     }
 
+    fn get_end_lsn(&self) -> Lsn {
+        let inner = self.inner.read().unwrap();
+        inner.drop_lsn.unwrap_or(Lsn(u64::MAX))
+    }
+
     fn path(&self) -> Option<PathBuf> {
         None
     }
@@ -134,22 +137,8 @@ impl Layer for InMemoryLayer {
         self.start_lsn
     }
 
-    fn get_end_lsn(&self) -> Lsn {
-        if let Some(end_lsn) = self.end_lsn {
-            return Lsn(end_lsn.0 + 1);
-        }
-
-        let inner = self.inner.lock().unwrap();
-
-        if let Some(drop_lsn) = inner.drop_lsn {
-            drop_lsn
-        } else {
-            Lsn(u64::MAX)
-        }
-    }
-
     fn is_dropped(&self) -> bool {
-        let inner = self.inner.lock().unwrap();
+        let inner = self.inner.read().unwrap();
         inner.drop_lsn.is_some()
     }
 
@@ -167,7 +156,7 @@ impl Layer for InMemoryLayer {
         let predecessor: Option<Arc<dyn Layer>>;
 
         {
-            let inner = self.inner.lock().unwrap();
+            let inner = self.inner.read().unwrap();
 
             // Scan the BTreeMap backwards, starting from reconstruct_data.lsn.
             let minkey = (blknum, Lsn(0));
@@ -214,13 +203,13 @@ impl Layer for InMemoryLayer {
     fn get_seg_size(&self, lsn: Lsn) -> Result<u32> {
         assert!(lsn >= self.start_lsn);
 
-        let inner = self.inner.lock().unwrap();
+        let inner = self.inner.read().unwrap();
         Ok(inner.get_seg_size(lsn))
     }
 
     /// Does this segment exist at given LSN?
     fn get_seg_exists(&self, lsn: Lsn) -> Result<bool> {
-        let inner = self.inner.lock().unwrap();
+        let inner = self.inner.read().unwrap();
 
         // If the segment created after requested LSN,
         // it doesn't exist in the layer. But we shouldn't
@@ -252,13 +241,12 @@ impl Layer for InMemoryLayer {
     }
 
     fn is_incremental(&self) -> bool {
-        let inner = self.inner.lock().unwrap();
-        inner.predecessor.is_some()
+        true
     }
 
     /// debugging function to print out the contents of the layer
     fn dump(&self) -> Result<()> {
-        let inner = self.inner.lock().unwrap();
+        let inner = self.inner.read().unwrap();
 
         let end_str = inner
             .drop_lsn
@@ -296,19 +284,7 @@ pub struct NonWriteableError;
 
 pub type WriteResult<T> = std::result::Result<T, NonWriteableError>;
 
-/// Helper struct to cleanup `InMemoryLayer::freeze` return signature.
-pub struct FreezeLayers {
-    /// Replacement layer for the layer which freeze was called on.
-    pub frozen: Arc<InMemoryLayer>,
-    /// New open layer containing leftover data.
-    pub open: Option<Arc<InMemoryLayer>>,
-}
-
 impl InMemoryLayer {
-    fn assert_not_frozen(&self) {
-        assert!(self.end_lsn.is_none());
-    }
-
     /// Return the oldest page version that's stored in this layer
     pub fn get_oldest_pending_lsn(&self) -> Lsn {
         self.oldest_pending_lsn
@@ -338,10 +314,10 @@ impl InMemoryLayer {
             tenantid,
             seg,
             start_lsn,
-            end_lsn: None,
             oldest_pending_lsn,
-            inner: Mutex::new(InMemoryLayerInner {
+            inner: RwLock::new(InMemoryLayerInner {
                 drop_lsn: None,
+                end_lsn: None,
                 page_versions: BTreeMap::new(),
                 segsizes: BTreeMap::new(),
                 writeable: true,
@@ -379,7 +355,6 @@ impl InMemoryLayer {
     /// Common subroutine of the public put_wal_record() and put_page_image() functions.
     /// Adds the page version to the in-memory tree
     pub fn put_page_version(&self, blknum: u32, lsn: Lsn, pv: PageVersion) -> WriteResult<u32> {
-        self.assert_not_frozen();
         assert!(self.seg.blknum_in_seg(blknum));
 
         trace!(
@@ -389,9 +364,10 @@ impl InMemoryLayer {
             self.timelineid,
             lsn
         );
-        let mut inner = self.inner.lock().unwrap();
+        let mut inner = self.inner.write().unwrap();
 
         inner.check_writeable()?;
+        inner.end_lsn = Some(lsn);
 
         let old = inner.page_versions.insert((blknum, lsn), pv);
 
@@ -407,7 +383,7 @@ impl InMemoryLayer {
         if self.seg.rel.is_blocky() {
             let newsize = blknum - self.seg.segno * RELISH_SEG_SIZE + 1;
 
-            // use inner get_seg_size, since calling self.get_seg_size will try to acquire self.inner.lock
+            // use inner get_seg_size, since calling self.get_seg_size will try to acquire the lock,
             // which we've just acquired above
             let oldsize = inner.get_seg_size(lsn);
             if newsize > oldsize {
@@ -458,10 +434,9 @@ impl InMemoryLayer {
 
     /// Remember that the relation was truncated at given LSN
     pub fn put_truncation(&self, lsn: Lsn, segsize: u32) -> WriteResult<()> {
-        self.assert_not_frozen();
-
-        let mut inner = self.inner.lock().unwrap();
+        let mut inner = self.inner.write().unwrap();
         inner.check_writeable()?;
+        inner.end_lsn = Some(lsn);
 
         // check that this we truncate to a smaller size than segment was before the truncation
         let oldsize = inner.get_seg_size(lsn);
@@ -479,9 +454,7 @@ impl InMemoryLayer {
 
     /// Remember that the segment was dropped at given LSN
     pub fn drop_segment(&self, lsn: Lsn) -> WriteResult<()> {
-        self.assert_not_frozen();
-
-        let mut inner = self.inner.lock().unwrap();
+        let mut inner = self.inner.write().unwrap();
 
         inner.check_writeable()?;
 
@@ -531,10 +504,10 @@ impl InMemoryLayer {
             tenantid,
             seg,
             start_lsn,
-            end_lsn: None,
             oldest_pending_lsn,
-            inner: Mutex::new(InMemoryLayerInner {
+            inner: RwLock::new(InMemoryLayerInner {
                 drop_lsn: None,
+                end_lsn: None,
                 page_versions: BTreeMap::new(),
                 segsizes,
                 writeable: true,
@@ -544,165 +517,58 @@ impl InMemoryLayer {
     }
 
     pub fn is_writeable(&self) -> bool {
-        let inner = self.inner.lock().unwrap();
+        let inner = self.inner.read().unwrap();
         inner.writeable
     }
 
-    /// Splits `self` into two InMemoryLayers: `frozen` and `open`.
-    /// All data up to and including `cutoff_lsn`
-    /// is copied to `frozen`, while the remaining data is copied to `open`.
-    /// After completion, self is non-writeable, but not frozen.
-    pub fn freeze(self: Arc<Self>, cutoff_lsn: Lsn) -> Result<FreezeLayers> {
-        info!(
-            "freezing in memory layer {} on timeline {} at {} (oldest {})",
-            self.filename().display(),
-            self.timelineid,
-            cutoff_lsn,
-            self.oldest_pending_lsn
-        );
-
-        self.assert_not_frozen();
-
-        let self_ref = self.clone();
-        let mut inner = self_ref.inner.lock().unwrap();
-        // Dropped layers don't need any special freeze actions,
-        // they are marked as non-writeable at drop and just
-        // written out to disk by checkpointer.
-        if inner.drop_lsn.is_some() {
-            assert!(!inner.writeable);
-            info!(
-                "freezing in memory layer for {} on timeline {} is dropped at {}",
-                self.seg,
-                self.timelineid,
-                inner.drop_lsn.unwrap()
-            );
-
-            // There should be no newer layer that refers this non-writeable layer,
-            // because layer that is created after dropped one represents a new rel.
-            return Ok(FreezeLayers {
-                frozen: self,
-                open: None,
-            });
-        }
-        assert!(inner.writeable);
-        inner.writeable = false;
-
-        // Divide all the page versions into old and new
-        // at the 'cutoff_lsn' point.
-        let mut before_segsizes = BTreeMap::new();
-        let mut after_segsizes = BTreeMap::new();
-        let mut after_oldest_lsn: Accum<Lsn> = Accum(None);
-        for (lsn, size) in inner.segsizes.iter() {
-            if *lsn > cutoff_lsn {
-                after_segsizes.insert(*lsn, *size);
-                after_oldest_lsn.accum(min, *lsn);
-            } else {
-                before_segsizes.insert(*lsn, *size);
-            }
-        }
-
-        let mut before_page_versions = BTreeMap::new();
-        let mut after_page_versions = BTreeMap::new();
-        for ((blknum, lsn), pv) in inner.page_versions.iter() {
-            if *lsn > cutoff_lsn {
-                after_page_versions.insert((*blknum, *lsn), pv.clone());
-                after_oldest_lsn.accum(min, *lsn);
-            } else {
-                before_page_versions.insert((*blknum, *lsn), pv.clone());
-            }
-        }
-
-        let frozen = Arc::new(InMemoryLayer {
-            conf: self.conf,
-            tenantid: self.tenantid,
-            timelineid: self.timelineid,
-            seg: self.seg,
-            start_lsn: self.start_lsn,
-            end_lsn: Some(cutoff_lsn),
-            oldest_pending_lsn: self.start_lsn,
-            inner: Mutex::new(InMemoryLayerInner {
-                drop_lsn: inner.drop_lsn,
-                page_versions: before_page_versions,
-                segsizes: before_segsizes,
-                writeable: false,
-                predecessor: inner.predecessor.clone(),
-            }),
-        });
-
-        let open = if !after_segsizes.is_empty() || !after_page_versions.is_empty() {
-            let mut new_open = Self::create_successor_layer(
-                self.conf,
-                frozen.clone(),
-                self.timelineid,
-                self.tenantid,
-                cutoff_lsn + 1,
-                after_oldest_lsn.0.unwrap(),
-            )?;
-
-            let new_inner = new_open.inner.get_mut().unwrap();
-            new_inner.page_versions.append(&mut after_page_versions);
-            new_inner.segsizes.append(&mut after_segsizes);
-
-            Some(Arc::new(new_open))
-        } else {
-            None
-        };
-
-        Ok(FreezeLayers { frozen, open })
-    }
-
     /// Write the this frozen in-memory layer to disk.
     ///
-    /// Returns new layers that replace this one.
-    /// If not dropped, returns a new image layer containing the page versions
-    /// at the `end_lsn`. Can also return a DeltaLayer that includes all the
-    /// WAL records between start and end LSN. (The delta layer is not needed
-    /// when a new relish is created with a single LSN, so that the start and
-    /// end LSN are the same.)
-    pub fn write_to_disk(&self, timeline: &LayeredTimeline) -> Result<Vec<Arc<dyn Layer>>> {
+    /// Returns new DeltaLayer that includes all the
+    /// WAL records between start and end LSN.
+    ///
+    pub fn write_to_disk(&self, timeline: &LayeredTimeline) -> Result<Arc<dyn Layer>> {
+        // Grab the lock in read-mode. We hold it over the I/O, but because this
+        // layer is not writeable anymore, no one should be trying to aquire the
+        // write lock on it, so we shouldn't block anyone. There's one exception
+        // though: another thread might have grabbed a reference to this layer
+        // in `get_layer_for_write' just before the checkpointer called
+        // `freeze`, and then `write_to_disk` on it. When the thread gets the
+        // lock, it will see that it's not writeable anymore and retry, but it
+        // would have to wait until we release it. That race condition is very
+        // rare though, so we just accept the potential latency hit for now.
+        let inner = self.inner.read().unwrap();
+
         trace!(
-            "write_to_disk {} end_lsn is {} get_end_lsn is {}",
+            "write_to_disk {} end_lsn is {}",
             self.filename().display(),
-            self.end_lsn.unwrap_or(Lsn(0)),
-            self.get_end_lsn()
+            inner.end_lsn.unwrap_or(Lsn(0)),
         );
 
-        let inner = self.inner.lock().unwrap();
-        let drop_lsn = inner.drop_lsn;
-
-        assert!(!inner.writeable);
-
-        let end_lsn = match drop_lsn {
-            Some(dlsn) => dlsn,
-            None => self.end_lsn.unwrap(),
-        };
-
         let predecessor = inner.predecessor.clone();
 
-        let mut before_page_versions;
-        let mut before_segsizes;
-        if inner.drop_lsn.is_none() {
-            before_segsizes = BTreeMap::new();
-            for (lsn, size) in inner.segsizes.iter() {
-                if *lsn <= end_lsn {
-                    before_segsizes.insert(*lsn, *size);
-                }
-            }
-
-            before_page_versions = BTreeMap::new();
-            for ((blknum, lsn), pv) in inner.page_versions.iter() {
-                if *lsn < end_lsn {
-                    before_page_versions.insert((*blknum, *lsn), pv.clone());
-                }
-            }
-        } else {
-            before_page_versions = inner.page_versions.clone();
-            before_segsizes = inner.segsizes.clone();
+        if let Some(drop_lsn) = inner.drop_lsn {
+            let delta_layer = DeltaLayer::create(
+                self.conf,
+                self.timelineid,
+                self.tenantid,
+                self.seg,
+                self.start_lsn,
+                drop_lsn,
+                true,
+                predecessor,
+                inner.page_versions.iter(),
+                inner.segsizes.clone(),
+            )?;
+            trace!(
+                "freeze: created delta layer for dropped segment {} {}-{}",
+                self.seg,
+                self.start_lsn,
+                drop_lsn
+            );
+            return Ok(Arc::new(delta_layer));
         }
 
-        drop(inner);
-
-        let mut frozen_layers: Vec<Arc<dyn Layer>> = Vec::new();
+        let end_lsn = inner.end_lsn.unwrap();
 
         if self.start_lsn != end_lsn {
             // Write the page versions before the cutoff to disk.
@@ -713,34 +579,15 @@ impl InMemoryLayer {
                 self.seg,
                 self.start_lsn,
                 end_lsn,
-                drop_lsn.is_some(),
+                false,
                 predecessor,
-                before_page_versions,
-                before_segsizes,
+                inner.page_versions.iter(),
+                inner.segsizes.clone(),
             )?;
-            frozen_layers.push(Arc::new(delta_layer));
-            trace!(
-                "freeze: created delta layer {} {}-{}",
-                self.seg,
-                self.start_lsn,
-                end_lsn
-            );
+            Ok(Arc::new(delta_layer))
         } else {
-            assert!(before_page_versions.is_empty());
-        }
-
-        if drop_lsn.is_none() {
-            // Write a new base image layer at the cutoff point
             let image_layer = ImageLayer::create_from_src(self.conf, timeline, self, end_lsn)?;
-            frozen_layers.push(Arc::new(image_layer));
-            trace!("freeze: created image layer {} at {}", self.seg, end_lsn);
+            Ok(Arc::new(image_layer))
         }
-
-        Ok(frozen_layers)
-    }
-
-    pub fn update_predecessor(&self, predecessor: Arc<dyn Layer>) -> Option<Arc<dyn Layer>> {
-        let mut inner = self.inner.lock().unwrap();
-        inner.predecessor.replace(predecessor)
     }
 }

pageserver/src/layered_repository/interval_tree.rs

@@ -133,11 +133,7 @@ where
                 found_start_point = true;
                 // It is an error to insert the same item to the tree twice.
                 assert!(
-                    point
-                        .elements
-                        .iter()
-                        .find(|x| Arc::ptr_eq(x, &item))
-                        .is_none(),
+                    !point.elements.iter().any(|x| Arc::ptr_eq(x, &item)),
                     "interval is already in the tree"
                 );
             }
@@ -180,7 +176,7 @@ where
                 found_start_point = true;
             }
             let len_before = point.elements.len();
-            point.elements.retain(|other| !Arc::ptr_eq(other, &item));
+            point.elements.retain(|other| !Arc::ptr_eq(other, item));
             let len_after = point.elements.len();
             assert_eq!(len_after + 1, len_before);
             if len_after == 0 {
@@ -223,7 +219,7 @@ where
                 }
             }
             // No more elements at this point. Move to next point.
-            if let Some((point_key, point)) = self.point_iter.next() {
+            if let Some((point_key, point)) = self.point_iter.next_back() {
                 self.elem_iter = Some((*point_key, point.elements.iter()));
                 continue;
             } else {
@@ -287,6 +283,7 @@ mod tests {
             write!(f, "{}", self.val)
         }
     }
+    #[rustfmt::skip]
     fn assert_search(
         tree: &IntervalTree<MockItem>,
         key: u32,
@@ -295,24 +292,20 @@ mod tests {
         if let Some(v) = tree.search(key) {
             let vstr = v.to_string();
 
-            if expected.is_empty() {
-                panic!("search with {} returned {}, expected None", key, v);
-            }
+            assert!(!expected.is_empty(), "search with {} returned {}, expected None", key, v);
+            assert!(
+                expected.contains(&vstr.as_str()),
+                "search with {} returned {}, expected one of: {:?}",
+                key, v, expected,
+            );
 
-            if !expected.contains(&vstr.as_str()) {
-                panic!(
-                    "search with {} returned {}, expected one of: {:?}",
-                    key, v, expected
-                );
-            }
             Some(v)
         } else {
-            if !expected.is_empty() {
-                panic!(
-                    "search with {} returned None, expected one of {:?}",
-                    key, expected
-                );
-            }
+            assert!(
+                expected.is_empty(),
+                "search with {} returned None, expected one of {:?}",
+                key, expected
+            );
             None
         }
     }

pageserver/src/lib.rs

@@ -13,7 +13,7 @@ pub mod http;
 pub mod layered_repository;
 pub mod page_service;
 pub mod relish;
-mod relish_storage;
+pub mod relish_storage;
 pub mod repository;
 pub mod restore_local_repo;
 pub mod tenant_mgr;
@@ -22,23 +22,25 @@ pub mod walreceiver;
 pub mod walredo;
 
 pub mod defaults {
+    use const_format::formatcp;
     use std::time::Duration;
 
     pub const DEFAULT_PG_LISTEN_PORT: u16 = 64000;
-    pub const DEFAULT_PG_LISTEN_ADDR: &str = "127.0.0.1:64000"; // can't format! const yet...
+    pub const DEFAULT_PG_LISTEN_ADDR: &str = formatcp!("127.0.0.1:{DEFAULT_PG_LISTEN_PORT}");
     pub const DEFAULT_HTTP_LISTEN_PORT: u16 = 9898;
-    pub const DEFAULT_HTTP_LISTEN_ADDR: &str = "127.0.0.1:9898";
+    pub const DEFAULT_HTTP_LISTEN_ADDR: &str = formatcp!("127.0.0.1:{DEFAULT_HTTP_LISTEN_PORT}");
 
     // FIXME: This current value is very low. I would imagine something like 1 GB or 10 GB
     // would be more appropriate. But a low value forces the code to be exercised more,
     // which is good for now to trigger bugs.
-    pub const DEFAULT_CHECKPOINT_DISTANCE: u64 = 64 * 1024 * 1024;
-    pub const DEFAULT_CHECKPOINT_PERIOD: Duration = Duration::from_secs(100);
+    pub const DEFAULT_CHECKPOINT_DISTANCE: u64 = 256 * 1024 * 1024;
+    pub const DEFAULT_CHECKPOINT_PERIOD: Duration = Duration::from_secs(1);
 
-    pub const DEFAULT_GC_HORIZON: u64 = 64 * 1024 * 1024;
-    pub const DEFAULT_GC_PERIOD: Duration = Duration::from_secs(100);
+    pub const DEFAULT_GC_HORIZON: u64 = 64u64 * 1024 * 1024 * 1024;
+    pub const DEFAULT_GC_PERIOD: Duration = Duration::from_secs(1);
 
     pub const DEFAULT_SUPERUSER: &str = "zenith_admin";
+    pub const DEFAULT_RELISH_STORAGE_MAX_CONCURRENT_SYNC_LIMITS: usize = 100;
 }
 
 lazy_static! {
@@ -153,8 +155,8 @@ impl PageServerConf {
             checkpoint_period: Duration::from_secs(10),
             gc_horizon: defaults::DEFAULT_GC_HORIZON,
             gc_period: Duration::from_secs(10),
-            listen_pg_addr: "127.0.0.1:5430".to_string(),
-            listen_http_addr: "127.0.0.1:9898".to_string(),
+            listen_pg_addr: defaults::DEFAULT_PG_LISTEN_ADDR.to_string(),
+            listen_http_addr: defaults::DEFAULT_HTTP_LISTEN_ADDR.to_string(),
             superuser: "zenith_admin".to_string(),
             workdir: repo_dir,
             pg_distrib_dir: "".into(),
@@ -167,18 +169,37 @@ impl PageServerConf {
 
 /// External relish storage configuration, enough for creating a client for that storage.
 #[derive(Debug, Clone)]
-pub enum RelishStorageConfig {
-    /// Root folder to place all stored relish data into.
+pub struct RelishStorageConfig {
+    /// Limits the number of concurrent sync operations between pageserver and relish storage.
+    pub max_concurrent_sync: usize,
+    /// The storage connection configuration.
+    pub storage: RelishStorageKind,
+}
+
+/// A kind of a relish storage to connect to, with its connection configuration.
+#[derive(Debug, Clone)]
+pub enum RelishStorageKind {
+    /// Storage based on local file system.
+    /// Specify a root folder to place all stored relish data into.
     LocalFs(PathBuf),
+    /// AWS S3 based storage, storing all relishes into the root
+    /// of the S3 bucket from the config.
     AwsS3(S3Config),
 }
 
 /// AWS S3 bucket coordinates and access credentials to manage the bucket contents (read and write).
 #[derive(Clone)]
 pub struct S3Config {
+    /// Name of the bucket to connect to.
     pub bucket_name: String,
+    /// The region where the bucket is located at.
     pub bucket_region: String,
+    /// "Login" to use when connecting to bucket.
+    /// Can be empty for cases like AWS k8s IAM
+    /// where we can allow certain pods to connect
+    /// to the bucket directly without any credentials.
     pub access_key_id: Option<String>,
+    /// "Password" to use when connecting to bucket.
     pub secret_access_key: Option<String>,
 }
 

pageserver/src/page_service.rs

@@ -194,7 +194,7 @@ pub fn thread_main(
         let local_auth = auth.clone();
         thread::spawn(move || {
             if let Err(err) = page_service_conn_main(conf, local_auth, socket, auth_type) {
-                error!("error: {}", err);
+                error!("page server thread exiting with error: {:#}", err);
             }
         });
     }
@@ -293,7 +293,9 @@ impl PageServerHandler {
             };
 
             let response = response.unwrap_or_else(|e| {
-                error!("error reading relation or page version: {}", e);
+                // print the all details to the log with {:#}, but for the client the
+                // error message is enough
+                error!("error reading relation or page version: {:#}", e);
                 PagestreamBeMessage::Error(PagestreamErrorResponse {
                     message: e.to_string(),
                 })

pageserver/src/relish_storage.rs

@@ -8,14 +8,43 @@
 
 mod local_fs;
 mod rust_s3;
-/// A queue and the background machinery behind it to upload
-/// local page server layer files to external storage.
-pub mod storage_uploader;
+/// A queue-based storage with the background machinery behind it to synchronize
+/// local page server layer files with external storage.
+mod synced_storage;
 
 use std::path::Path;
+use std::thread;
 
 use anyhow::Context;
 
+use self::local_fs::LocalFs;
+pub use self::synced_storage::schedule_timeline_upload;
+use crate::relish_storage::rust_s3::RustS3;
+use crate::{PageServerConf, RelishStorageKind};
+
+pub fn run_storage_sync_thread(
+    config: &'static PageServerConf,
+) -> anyhow::Result<Option<thread::JoinHandle<anyhow::Result<()>>>> {
+    match &config.relish_storage_config {
+        Some(relish_storage_config) => {
+            let max_concurrent_sync = relish_storage_config.max_concurrent_sync;
+            match &relish_storage_config.storage {
+                RelishStorageKind::LocalFs(root) => synced_storage::run_storage_sync_thread(
+                    config,
+                    LocalFs::new(root.clone())?,
+                    max_concurrent_sync,
+                ),
+                RelishStorageKind::AwsS3(s3_config) => synced_storage::run_storage_sync_thread(
+                    config,
+                    RustS3::new(s3_config)?,
+                    max_concurrent_sync,
+                ),
+            }
+        }
+        None => Ok(None),
+    }
+}
+
 /// Storage (potentially remote) API to manage its state.
 #[async_trait::async_trait]
 pub trait RelishStorage: Send + Sync {

pageserver/src/relish_storage/storage_uploader.rs

@@ -1,116 +0,0 @@
-use std::{
-    collections::VecDeque,
-    path::{Path, PathBuf},
-    sync::{Arc, Mutex},
-    thread,
-};
-
-use zenith_utils::zid::ZTimelineId;
-
-use crate::{relish_storage::RelishStorage, RelishStorageConfig};
-
-use super::{local_fs::LocalFs, rust_s3::RustS3};
-
-pub struct QueueBasedRelishUploader {
-    upload_queue: Arc<Mutex<VecDeque<(ZTimelineId, PathBuf)>>>,
-}
-
-impl QueueBasedRelishUploader {
-    pub fn new(
-        config: &RelishStorageConfig,
-        page_server_workdir: &'static Path,
-    ) -> anyhow::Result<Self> {
-        let upload_queue = Arc::new(Mutex::new(VecDeque::new()));
-        let _handle = match config {
-            RelishStorageConfig::LocalFs(root) => {
-                let relish_storage = LocalFs::new(root.clone())?;
-                create_upload_thread(
-                    Arc::clone(&upload_queue),
-                    relish_storage,
-                    page_server_workdir,
-                )?
-            }
-            RelishStorageConfig::AwsS3(s3_config) => {
-                let relish_storage = RustS3::new(s3_config)?;
-                create_upload_thread(
-                    Arc::clone(&upload_queue),
-                    relish_storage,
-                    page_server_workdir,
-                )?
-            }
-        };
-
-        Ok(Self { upload_queue })
-    }
-
-    pub fn schedule_upload(&self, timeline_id: ZTimelineId, relish_path: PathBuf) {
-        self.upload_queue
-            .lock()
-            .unwrap()
-            .push_back((timeline_id, relish_path))
-    }
-}
-
-fn create_upload_thread<P, S: 'static + RelishStorage<RelishStoragePath = P>>(
-    upload_queue: Arc<Mutex<VecDeque<(ZTimelineId, PathBuf)>>>,
-    relish_storage: S,
-    page_server_workdir: &'static Path,
-) -> std::io::Result<thread::JoinHandle<()>> {
-    let runtime = tokio::runtime::Builder::new_current_thread()
-        .enable_all()
-        .build()?;
-    thread::Builder::new()
-        .name("Queue based relish uploader".to_string())
-        .spawn(move || loop {
-            runtime.block_on(async {
-                upload_loop_step(&upload_queue, &relish_storage, page_server_workdir).await;
-            })
-        })
-}
-
-async fn upload_loop_step<P, S: 'static + RelishStorage<RelishStoragePath = P>>(
-    upload_queue: &Mutex<VecDeque<(ZTimelineId, PathBuf)>>,
-    relish_storage: &S,
-    page_server_workdir: &Path,
-) {
-    let mut queue_accessor = upload_queue.lock().unwrap();
-    log::debug!("current upload queue length: {}", queue_accessor.len());
-    let next_upload = queue_accessor.pop_front();
-    drop(queue_accessor);
-
-    let (relish_timeline_id, relish_local_path) = match next_upload {
-        Some(data) => data,
-        None => {
-            // Don't spin and allow others to use the queue.
-            // In future, could be improved to be more clever about delays depending on relish upload stats
-            thread::sleep(std::time::Duration::from_secs(1));
-            return;
-        }
-    };
-
-    if let Err(e) = upload_relish(relish_storage, page_server_workdir, &relish_local_path).await {
-        log::error!(
-            "Failed to upload relish '{}' for timeline {}, reason: {}",
-            relish_local_path.display(),
-            relish_timeline_id,
-            e
-        );
-        upload_queue
-            .lock()
-            .unwrap()
-            .push_back((relish_timeline_id, relish_local_path))
-    } else {
-        log::debug!("Relish successfully uploaded");
-    }
-}
-
-async fn upload_relish<P, S: RelishStorage<RelishStoragePath = P>>(
-    relish_storage: &S,
-    page_server_workdir: &Path,
-    relish_local_path: &Path,
-) -> anyhow::Result<()> {
-    let destination = S::derive_destination(page_server_workdir, relish_local_path)?;
-    relish_storage
-        .upload_relish(relish_local_path, &destination)
-        .await
-}

pageserver/src/relish_storage/synced_storage.rs

@@ -0,0 +1,52 @@
+use std::time::Duration;
+use std::{collections::BinaryHeap, sync::Mutex, thread};
+
+use crate::{relish_storage::RelishStorage, PageServerConf};
+
+lazy_static::lazy_static! {
+    static ref UPLOAD_QUEUE: Mutex<BinaryHeap<SyncTask>> = Mutex::new(BinaryHeap::new());
+}
+
+pub fn schedule_timeline_upload(_local_timeline: ()) {
+    // UPLOAD_QUEUE
+    //     .lock()
+    //     .unwrap()
+    //     .push(SyncTask::Upload(local_timeline))
+}
+
+#[derive(Debug, PartialEq, Eq, PartialOrd, Ord)]
+enum SyncTask {}
+
+pub fn run_storage_sync_thread<
+    P: std::fmt::Debug,
+    S: 'static + RelishStorage<RelishStoragePath = P>,
+>(
+    config: &'static PageServerConf,
+    relish_storage: S,
+    max_concurrent_sync: usize,
+) -> anyhow::Result<Option<thread::JoinHandle<anyhow::Result<()>>>> {
+    let runtime = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()?;
+
+    let handle = thread::Builder::new()
+        .name("Queue based relish storage sync".to_string())
+        .spawn(move || loop {
+            let mut queue_accessor = UPLOAD_QUEUE.lock().unwrap();
+            log::debug!("Upload queue length: {}", queue_accessor.len());
+            let next_task = queue_accessor.pop();
+            drop(queue_accessor);
+            match next_task {
+                Some(task) => runtime.block_on(async {
+                    // suppress warnings
+                    let _ = (config, task, &relish_storage, max_concurrent_sync);
+                    todo!("omitted for brevity")
+                }),
+                None => {
+                    thread::sleep(Duration::from_secs(1));
+                    continue;
+                }
+            }
+        })?;
+    Ok(Some(handle))
+}

pageserver/src/repository.rs

@@ -212,12 +212,18 @@ mod tests {
     use crate::layered_repository::LayeredRepository;
     use crate::walredo::{WalRedoError, WalRedoManager};
     use crate::PageServerConf;
+    use hex_literal::hex;
     use postgres_ffi::pg_constants;
     use postgres_ffi::xlog_utils::SIZEOF_CHECKPOINT;
     use std::fs;
-    use std::str::FromStr;
+    use std::path::PathBuf;
     use zenith_utils::zid::ZTenantId;
 
+    const TIMELINE_ID: ZTimelineId =
+        ZTimelineId::from_array(hex!("11223344556677881122334455667788"));
+    const NEW_TIMELINE_ID: ZTimelineId =
+        ZTimelineId::from_array(hex!("AA223344556677881122334455667788"));
+
     /// Arbitrary relation tag, for testing.
     const TESTREL_A: RelishTag = RelishTag::Relation(RelTag {
         spcnode: 0,
@@ -253,39 +259,53 @@ mod tests {
     static ZERO_PAGE: Bytes = Bytes::from_static(&[0u8; 8192]);
     static ZERO_CHECKPOINT: Bytes = Bytes::from_static(&[0u8; SIZEOF_CHECKPOINT]);
 
-    fn get_test_repo(test_name: &str) -> Result<Box<dyn Repository>> {
-        let repo_dir = PageServerConf::test_repo_dir(test_name);
-        let _ = fs::remove_dir_all(&repo_dir);
-        fs::create_dir_all(&repo_dir)?;
-        fs::create_dir_all(&repo_dir.join("timelines"))?;
+    struct RepoHarness {
+        conf: &'static PageServerConf,
+        tenant_id: ZTenantId,
+    }
 
-        let conf = PageServerConf::dummy_conf(repo_dir);
-        // Make a static copy of the config. This can never be free'd, but that's
-        // OK in a test.
-        let conf: &'static PageServerConf = Box::leak(Box::new(conf));
-        let tenantid = ZTenantId::generate();
-        fs::create_dir_all(conf.tenant_path(&tenantid)).unwrap();
+    impl RepoHarness {
+        fn create(test_name: &'static str) -> Result<Self> {
+            let repo_dir = PageServerConf::test_repo_dir(test_name);
+            let _ = fs::remove_dir_all(&repo_dir);
+            fs::create_dir_all(&repo_dir)?;
+            fs::create_dir_all(&repo_dir.join("timelines"))?;
 
-        let walredo_mgr = TestRedoManager {};
+            let conf = PageServerConf::dummy_conf(repo_dir);
+            // Make a static copy of the config. This can never be free'd, but that's
+            // OK in a test.
+            let conf: &'static PageServerConf = Box::leak(Box::new(conf));
 
-        let repo = Box::new(LayeredRepository::new(
-            conf,
-            Arc::new(walredo_mgr),
-            tenantid,
-        ));
+            let tenant_id = ZTenantId::generate();
+            fs::create_dir_all(conf.tenant_path(&tenant_id))?;
 
-        Ok(repo)
+            Ok(Self { conf, tenant_id })
+        }
+
+        fn load(&self) -> Box<dyn Repository> {
+            let walredo_mgr = Arc::new(TestRedoManager);
+
+            Box::new(LayeredRepository::new(
+                self.conf,
+                walredo_mgr,
+                self.tenant_id,
+                false,
+            ))
+        }
+
+        fn timeline_path(&self, timeline_id: &ZTimelineId) -> PathBuf {
+            self.conf.timeline_path(timeline_id, &self.tenant_id)
+        }
     }
 
     #[test]
     fn test_relsize() -> Result<()> {
-        let repo = get_test_repo("test_relsize")?;
+        let repo = RepoHarness::create("test_relsize")?.load();
         // get_timeline() with non-existent timeline id should fail
         //repo.get_timeline("11223344556677881122334455667788");
 
         // Create timeline to work on
-        let timelineid = ZTimelineId::from_str("11223344556677881122334455667788").unwrap();
-        let tline = repo.create_empty_timeline(timelineid)?;
+        let tline = repo.create_empty_timeline(TIMELINE_ID)?;
 
         tline.put_page_image(TESTREL_A, 0, Lsn(0x20), TEST_IMG("foo blk 0 at 2"))?;
         tline.put_page_image(TESTREL_A, 0, Lsn(0x20), TEST_IMG("foo blk 0 at 2"))?;
@@ -399,11 +419,10 @@ mod tests {
     // and then created it again within the same layer.
     #[test]
     fn test_drop_extend() -> Result<()> {
-        let repo = get_test_repo("test_drop_extend")?;
+        let repo = RepoHarness::create("test_drop_extend")?.load();
 
         // Create timeline to work on
-        let timelineid = ZTimelineId::from_str("11223344556677881122334455667788").unwrap();
-        let tline = repo.create_empty_timeline(timelineid)?;
+        let tline = repo.create_empty_timeline(TIMELINE_ID)?;
 
         tline.put_page_image(TESTREL_A, 0, Lsn(0x20), TEST_IMG("foo blk 0 at 2"))?;
         tline.advance_last_record_lsn(Lsn(0x20));
@@ -436,11 +455,10 @@ mod tests {
     // and then extended it again within the same layer.
     #[test]
     fn test_truncate_extend() -> Result<()> {
-        let repo = get_test_repo("test_truncate_extend")?;
+        let repo = RepoHarness::create("test_truncate_extend")?.load();
 
         // Create timeline to work on
-        let timelineid = ZTimelineId::from_str("11223344556677881122334455667788").unwrap();
-        let tline = repo.create_empty_timeline(timelineid)?;
+        let tline = repo.create_empty_timeline(TIMELINE_ID)?;
 
         //from storage_layer.rs
         const RELISH_SEG_SIZE: u32 = 10 * 1024 * 1024 / 8192;
@@ -537,9 +555,8 @@ mod tests {
     /// split into multiple 1 GB segments in Postgres.
     #[test]
     fn test_large_rel() -> Result<()> {
-        let repo = get_test_repo("test_large_rel")?;
-        let timelineid = ZTimelineId::from_str("11223344556677881122334455667788").unwrap();
-        let tline = repo.create_empty_timeline(timelineid)?;
+        let repo = RepoHarness::create("test_large_rel")?.load();
+        let tline = repo.create_empty_timeline(TIMELINE_ID)?;
 
         let mut lsn = 0x10;
         for blknum in 0..pg_constants::RELSEG_SIZE + 1 {
@@ -600,9 +617,8 @@ mod tests {
     ///
     #[test]
     fn test_list_rels_drop() -> Result<()> {
-        let repo = get_test_repo("test_list_rels_drop")?;
-        let timelineid = ZTimelineId::from_str("11223344556677881122334455667788").unwrap();
-        let tline = repo.create_empty_timeline(timelineid)?;
+        let repo = RepoHarness::create("test_list_rels_drop")?.load();
+        let tline = repo.create_empty_timeline(TIMELINE_ID)?;
         const TESTDB: u32 = 111;
 
         // Import initial dummy checkpoint record, otherwise the get_timeline() call
@@ -620,9 +636,8 @@ mod tests {
         assert!(tline.list_rels(0, TESTDB, Lsn(0x30))?.contains(&TESTREL_A));
 
         // Create a branch, check that the relation is visible there
-        let newtimelineid = ZTimelineId::from_str("AA223344556677881122334455667788").unwrap();
-        repo.branch_timeline(timelineid, newtimelineid, Lsn(0x30))?;
-        let newtline = repo.get_timeline(newtimelineid)?;
+        repo.branch_timeline(TIMELINE_ID, NEW_TIMELINE_ID, Lsn(0x30))?;
+        let newtline = repo.get_timeline(NEW_TIMELINE_ID)?;
 
         assert!(newtline
             .list_rels(0, TESTDB, Lsn(0x30))?
@@ -642,7 +657,7 @@ mod tests {
 
         // Run checkpoint and garbage collection and check that it's still not visible
         newtline.checkpoint()?;
-        repo.gc_iteration(Some(newtimelineid), 0, true)?;
+        repo.gc_iteration(Some(NEW_TIMELINE_ID), 0, true)?;
 
         assert!(!newtline
             .list_rels(0, TESTDB, Lsn(0x40))?
@@ -656,9 +671,8 @@ mod tests {
     ///
     #[test]
     fn test_branch() -> Result<()> {
-        let repo = get_test_repo("test_branch")?;
-        let timelineid = ZTimelineId::from_str("11223344556677881122334455667788").unwrap();
-        let tline = repo.create_empty_timeline(timelineid)?;
+        let repo = RepoHarness::create("test_branch")?.load();
+        let tline = repo.create_empty_timeline(TIMELINE_ID)?;
 
         // Import initial dummy checkpoint record, otherwise the get_timeline() call
         // after branching fails below
@@ -676,9 +690,8 @@ mod tests {
         assert_current_logical_size(&tline, Lsn(0x40));
 
         // Branch the history, modify relation differently on the new timeline
-        let newtimelineid = ZTimelineId::from_str("AA223344556677881122334455667788").unwrap();
-        repo.branch_timeline(timelineid, newtimelineid, Lsn(0x30))?;
-        let newtline = repo.get_timeline(newtimelineid)?;
+        repo.branch_timeline(TIMELINE_ID, NEW_TIMELINE_ID, Lsn(0x30))?;
+        let newtline = repo.get_timeline(NEW_TIMELINE_ID)?;
 
         newtline.put_page_image(TESTREL_A, 0, Lsn(0x40), TEST_IMG("bar blk 0 at 4"))?;
         newtline.advance_last_record_lsn(Lsn(0x40));
@@ -706,8 +719,89 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn corrupt_metadata() -> Result<()> {
+        const TEST_NAME: &str = "corrupt_metadata";
+        let harness = RepoHarness::create(TEST_NAME)?;
+        let repo = harness.load();
+
+        repo.create_empty_timeline(TIMELINE_ID)?;
+        drop(repo);
+
+        let metadata_path = harness.timeline_path(&TIMELINE_ID).join("metadata");
+
+        assert!(metadata_path.is_file());
+
+        let mut metadata_bytes = std::fs::read(&metadata_path)?;
+        assert_eq!(metadata_bytes.len(), 512);
+        metadata_bytes[512 - 4 - 2] ^= 1;
+        std::fs::write(metadata_path, metadata_bytes)?;
+
+        let new_repo = harness.load();
+        let err = new_repo.get_timeline(TIMELINE_ID).err().unwrap();
+        assert!(err.to_string().contains("checksum"));
+
+        Ok(())
+    }
+
+    #[test]
+    fn future_layerfiles() -> Result<()> {
+        const TEST_NAME: &str = "future_layerfiles";
+        let harness = RepoHarness::create(TEST_NAME)?;
+        let repo = harness.load();
+
+        repo.create_empty_timeline(TIMELINE_ID)?;
+        drop(repo);
+
+        let timeline_path = harness.timeline_path(&TIMELINE_ID);
+
+        let make_empty_file = |filename: &str| -> std::io::Result<()> {
+            let path = timeline_path.join(filename);
+
+            assert!(!path.exists());
+            std::fs::write(&path, &[])?;
+
+            Ok(())
+        };
+
+        let image_filename = format!("pg_control_0_{:016X}", 8000);
+        let delta_filename = format!("pg_control_0_{:016X}_{:016X}", 8000, 8008);
+
+        make_empty_file(&image_filename)?;
+        make_empty_file(&delta_filename)?;
+
+        let new_repo = harness.load();
+        new_repo.get_timeline(TIMELINE_ID).unwrap();
+        drop(new_repo);
+
+        let check_old = |filename: &str, num: u32| {
+            let path = timeline_path.join(filename);
+            assert!(!path.exists());
+
+            let backup_path = timeline_path.join(format!("{}.{}.old", filename, num));
+            assert!(backup_path.exists());
+        };
+
+        check_old(&image_filename, 0);
+        check_old(&delta_filename, 0);
+
+        make_empty_file(&image_filename)?;
+        make_empty_file(&delta_filename)?;
+
+        let new_repo = harness.load();
+        new_repo.get_timeline(TIMELINE_ID).unwrap();
+        drop(new_repo);
+
+        check_old(&image_filename, 0);
+        check_old(&delta_filename, 0);
+        check_old(&image_filename, 1);
+        check_old(&delta_filename, 1);
+
+        Ok(())
+    }
+
     // Mock WAL redo manager that doesn't do much
-    struct TestRedoManager {}
+    struct TestRedoManager;
 
     impl WalRedoManager for TestRedoManager {
         fn request_redo(

pageserver/src/restore_local_repo.rs

@@ -9,11 +9,9 @@ use std::cmp::min;
 use std::fs;
 use std::fs::File;
 use std::io::Read;
-use std::io::Seek;
-use std::io::SeekFrom;
-use std::path::{Path, PathBuf};
+use std::path::Path;
 
-use anyhow::Result;
+use anyhow::{bail, Result};
 use bytes::{Buf, Bytes};
 
 use crate::relish::*;
@@ -175,8 +173,7 @@ fn import_relfile(
                     break;
                 }
                 _ => {
-                    error!("error reading file: {:?} ({})", path, e);
-                    break;
+                    bail!("error reading file {}: {:#}", path.display(), e);
                 }
             },
         };
@@ -270,8 +267,7 @@ fn import_slru_file(timeline: &dyn Timeline, lsn: Lsn, slru: SlruKind, path: &Pa
                     break;
                 }
                 _ => {
-                    error!("error reading file: {:?} ({})", path, e);
-                    break;
+                    bail!("error reading file {}: {:#}", path.display(), e);
                 }
             },
         };
@@ -283,100 +279,6 @@ fn import_slru_file(timeline: &dyn Timeline, lsn: Lsn, slru: SlruKind, path: &Pa
     Ok(())
 }
 
-/// Scan PostgreSQL WAL files in given directory
-/// and load all records >= 'startpoint' into the repository.
-pub fn import_timeline_wal(walpath: &Path, timeline: &dyn Timeline, startpoint: Lsn) -> Result<()> {
-    let mut waldecoder = WalStreamDecoder::new(startpoint);
-
-    let mut segno = startpoint.segment_number(pg_constants::WAL_SEGMENT_SIZE);
-    let mut offset = startpoint.segment_offset(pg_constants::WAL_SEGMENT_SIZE);
-    let mut last_lsn = startpoint;
-
-    let checkpoint_bytes = timeline.get_page_at_lsn(RelishTag::Checkpoint, 0, startpoint)?;
-    let mut checkpoint = CheckPoint::decode(&checkpoint_bytes)?;
-
-    loop {
-        // FIXME: assume postgresql tli 1 for now
-        let filename = XLogFileName(1, segno, pg_constants::WAL_SEGMENT_SIZE);
-        let mut buf = Vec::new();
-
-        //Read local file
-        let mut path = walpath.join(&filename);
-
-        // It could be as .partial
-        if !PathBuf::from(&path).exists() {
-            path = walpath.join(filename + ".partial");
-        }
-
-        // Slurp the WAL file
-        let open_result = File::open(&path);
-        if let Err(e) = &open_result {
-            if e.kind() == std::io::ErrorKind::NotFound {
-                break;
-            }
-        }
-        let mut file = open_result?;
-
-        if offset > 0 {
-            file.seek(SeekFrom::Start(offset as u64))?;
-        }
-
-        let nread = file.read_to_end(&mut buf)?;
-        if nread != pg_constants::WAL_SEGMENT_SIZE - offset as usize {
-            // Maybe allow this for .partial files?
-            error!("read only {} bytes from WAL file", nread);
-        }
-
-        waldecoder.feed_bytes(&buf);
-
-        let mut nrecords = 0;
-        loop {
-            let rec = waldecoder.poll_decode();
-            if rec.is_err() {
-                // Assume that an error means we've reached the end of
-                // a partial WAL record. So that's ok.
-                trace!("WAL decoder error {:?}", rec);
-                break;
-            }
-            if let Some((lsn, recdata)) = rec.unwrap() {
-                // The previous record has been handled, let the repository know that
-                // it is up-to-date to this LSN. (We do this here on the "next" iteration,
-                // rather than right after the save_decoded_record, because at the end of
-                // the WAL, we will also need to perform the update of the checkpoint data
-                // with the same LSN as the last actual record.)
-                timeline.advance_last_record_lsn(last_lsn);
-
-                let decoded = decode_wal_record(recdata.clone());
-                save_decoded_record(&mut checkpoint, timeline, &decoded, recdata, lsn)?;
-                last_lsn = lsn;
-            } else {
-                break;
-            }
-            nrecords += 1;
-        }
-
-        info!("imported {} records up to {}", nrecords, last_lsn);
-
-        segno += 1;
-        offset = 0;
-    }
-
-    if last_lsn != startpoint {
-        info!(
-            "reached end of WAL at {}, updating checkpoint info",
-            last_lsn
-        );
-        let checkpoint_bytes = checkpoint.encode();
-        timeline.put_page_image(RelishTag::Checkpoint, 0, last_lsn, checkpoint_bytes)?;
-
-        timeline.advance_last_record_lsn(last_lsn);
-    } else {
-        info!("no WAL to import at {}", last_lsn);
-    }
-
-    Ok(())
-}
-
 ///
 /// Helper function to parse a WAL record and call the Timeline's PUT functions for all the
 /// relations/pages that the record affects.

pageserver/src/tenant_mgr.rs

@@ -9,46 +9,84 @@ use crate::PageServerConf;
 use anyhow::{anyhow, bail, Context, Result};
 use lazy_static::lazy_static;
 use log::info;
+use std::collections::hash_map::Entry;
 use std::collections::HashMap;
 use std::fs;
 use std::str::FromStr;
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Mutex, MutexGuard};
 use zenith_utils::zid::{ZTenantId, ZTimelineId};
 
 lazy_static! {
-    pub static ref REPOSITORY: Mutex<HashMap<ZTenantId, Arc<dyn Repository>>> =
+    static ref REPOSITORY: Mutex<HashMap<ZTenantId, Arc<dyn Repository>>> =
         Mutex::new(HashMap::new());
 }
 
-pub fn init(conf: &'static PageServerConf) {
-    let mut m = REPOSITORY.lock().unwrap();
+fn access_repository() -> MutexGuard<'static, HashMap<ZTenantId, Arc<dyn Repository>>> {
+    REPOSITORY.lock().unwrap()
+}
 
+pub fn init(conf: &'static PageServerConf) {
+    let mut m = access_repository();
     for dir_entry in fs::read_dir(conf.tenants_path()).unwrap() {
         let tenantid =
             ZTenantId::from_str(dir_entry.unwrap().file_name().to_str().unwrap()).unwrap();
-
-        // Set up a WAL redo manager, for applying WAL records.
-        let walredo_mgr = PostgresRedoManager::new(conf, tenantid);
-
-        // Set up an object repository, for actual data storage.
-        let repo = Arc::new(LayeredRepository::new(
-            conf,
-            Arc::new(walredo_mgr),
-            tenantid,
-        ));
-        LayeredRepository::launch_checkpointer_thread(conf, repo.clone());
-        LayeredRepository::launch_gc_thread(conf, repo.clone());
-
+        let repo = init_repo(conf, tenantid);
         info!("initialized storage for tenant: {}", &tenantid);
         m.insert(tenantid, repo);
     }
 }
 
+fn init_repo(conf: &'static PageServerConf, tenant_id: ZTenantId) -> Arc<LayeredRepository> {
+    // Set up a WAL redo manager, for applying WAL records.
+    let walredo_mgr = PostgresRedoManager::new(conf, tenant_id);
+
+    // Set up an object repository, for actual data storage.
+    let repo = Arc::new(LayeredRepository::new(
+        conf,
+        Arc::new(walredo_mgr),
+        tenant_id,
+        true,
+    ));
+    LayeredRepository::launch_checkpointer_thread(conf, repo.clone());
+    LayeredRepository::launch_gc_thread(conf, repo.clone());
+    repo
+}
+
+// TODO kb Currently unused function, will later be used when the relish storage downloads a new layer.
+// Relevant PR: https://github.com/zenithdb/zenith/pull/686
+pub fn register_relish_download(
+    conf: &'static PageServerConf,
+    tenant_id: ZTenantId,
+    timeline_id: ZTimelineId,
+) {
+    log::info!(
+        "Registering new download, tenant id {}, timeline id: {}",
+        tenant_id,
+        timeline_id
+    );
+    match access_repository().entry(tenant_id) {
+        Entry::Occupied(o) => init_timeline(o.get().as_ref(), timeline_id),
+        Entry::Vacant(v) => {
+            log::info!("New repo initialized");
+            let new_repo = init_repo(conf, tenant_id);
+            init_timeline(new_repo.as_ref(), timeline_id);
+            v.insert(new_repo);
+        }
+    }
+}
+
+fn init_timeline(repo: &dyn Repository, timeline_id: ZTimelineId) {
+    match repo.get_timeline(timeline_id) {
+        Ok(_timeline) => log::info!("Successfully initialized timeline {}", timeline_id),
+        Err(e) => log::error!("Failed to init timeline {}, reason: {:#}", timeline_id, e),
+    }
+}
+
 pub fn create_repository_for_tenant(
     conf: &'static PageServerConf,
     tenantid: ZTenantId,
 ) -> Result<()> {
-    let mut m = REPOSITORY.lock().unwrap();
+    let mut m = access_repository();
 
     // First check that the tenant doesn't exist already
     if m.get(&tenantid).is_some() {
@@ -63,14 +101,13 @@ pub fn create_repository_for_tenant(
 }
 
 pub fn insert_repository_for_tenant(tenantid: ZTenantId, repo: Arc<dyn Repository>) {
-    let o = &mut REPOSITORY.lock().unwrap();
-    o.insert(tenantid, repo);
+    access_repository().insert(tenantid, repo);
 }
 
 pub fn get_repository_for_tenant(tenantid: ZTenantId) -> Result<Arc<dyn Repository>> {
-    let o = &REPOSITORY.lock().unwrap();
-    o.get(&tenantid)
-        .map(|repo| Arc::clone(repo))
+    access_repository()
+        .get(&tenantid)
+        .map(Arc::clone)
         .ok_or_else(|| anyhow!("repository not found for tenant name {}", tenantid))
 }
 

pageserver/src/walreceiver.rs

@@ -10,7 +10,7 @@ use crate::restore_local_repo;
 use crate::tenant_mgr;
 use crate::waldecoder::*;
 use crate::PageServerConf;
-use anyhow::{Error, Result};
+use anyhow::{bail, Error, Result};
 use lazy_static::lazy_static;
 use log::*;
 use postgres::fallible_iterator::FallibleIterator;
@@ -20,15 +20,15 @@ use postgres_ffi::xlog_utils::*;
 use postgres_ffi::*;
 use postgres_protocol::message::backend::ReplicationMessage;
 use postgres_types::PgLsn;
+use std::cell::Cell;
 use std::cmp::{max, min};
 use std::collections::HashMap;
 use std::fs;
-use std::fs::{File, OpenOptions};
-use std::io::{Seek, SeekFrom, Write};
 use std::str::FromStr;
 use std::sync::Mutex;
 use std::thread;
 use std::thread::sleep;
+use std::thread_local;
 use std::time::{Duration, SystemTime};
 use zenith_utils::lsn::Lsn;
 use zenith_utils::zid::ZTenantId;
@@ -46,6 +46,13 @@ lazy_static! {
         Mutex::new(HashMap::new());
 }
 
+thread_local! {
+    // Boolean that is true only for WAL receiver threads
+    //
+    // This is used in `wait_lsn` to guard against usage that might lead to a deadlock.
+    pub(crate) static IS_WAL_RECEIVER: Cell<bool> = Cell::new(false);
+}
+
 // Launch a new WAL receiver, or tell one that's running about change in connection string
 pub fn launch_wal_receiver(
     conf: &'static PageServerConf,
@@ -66,12 +73,10 @@ pub fn launch_wal_receiver(
             receivers.insert(timelineid, receiver);
 
             // Also launch a new thread to handle this connection
-            //
-            // NOTE: This thread name is checked in the assertion in wait_lsn. If you change
-            // this, make sure you update the assertion too.
             let _walreceiver_thread = thread::Builder::new()
                 .name("WAL receiver thread".into())
                 .spawn(move || {
+                    IS_WAL_RECEIVER.with(|c| c.set(true));
                     thread_main(conf, timelineid, tenantid);
                 })
                 .unwrap();
@@ -160,7 +165,7 @@ fn walreceiver_main(
     let mut startpoint = last_rec_lsn;
 
     if startpoint == Lsn(0) {
-        error!("No previous WAL position");
+        bail!("No previous WAL position");
     }
 
     // There might be some padding after the last full record, skip it.
@@ -192,15 +197,6 @@ fn walreceiver_main(
                 let endlsn = startlsn + data.len() as u64;
                 let prev_last_rec_lsn = last_rec_lsn;
 
-                write_wal_file(
-                    conf,
-                    startlsn,
-                    &timelineid,
-                    pg_constants::WAL_SEGMENT_SIZE,
-                    data,
-                    &tenantid,
-                )?;
-
                 trace!("received XLogData between {} and {}", startlsn, endlsn);
 
                 waldecoder.feed_bytes(data);
@@ -403,98 +399,3 @@ pub fn identify_system(client: &mut Client) -> Result<IdentifySystem, Error> {
         Err(IdentifyError.into())
     }
 }
-
-fn write_wal_file(
-    conf: &PageServerConf,
-    startpos: Lsn,
-    timelineid: &ZTimelineId,
-    wal_seg_size: usize,
-    buf: &[u8],
-    tenantid: &ZTenantId,
-) -> anyhow::Result<()> {
-    let mut bytes_left: usize = buf.len();
-    let mut bytes_written: usize = 0;
-    let mut partial;
-    let mut start_pos = startpos;
-    const ZERO_BLOCK: &[u8] = &[0u8; XLOG_BLCKSZ];
-
-    let wal_dir = conf.wal_dir_path(timelineid, tenantid);
-
-    /* Extract WAL location for this block */
-    let mut xlogoff = start_pos.segment_offset(wal_seg_size);
-
-    while bytes_left != 0 {
-        let bytes_to_write;
-
-        /*
-         * If crossing a WAL boundary, only write up until we reach wal
-         * segment size.
-         */
-        if xlogoff + bytes_left > wal_seg_size {
-            bytes_to_write = wal_seg_size - xlogoff;
-        } else {
-            bytes_to_write = bytes_left;
-        }
-
-        /* Open file */
-        let segno = start_pos.segment_number(wal_seg_size);
-        let wal_file_name = XLogFileName(
-            1, // FIXME: always use Postgres timeline 1
-            segno,
-            wal_seg_size,
-        );
-        let wal_file_path = wal_dir.join(wal_file_name.clone());
-        let wal_file_partial_path = wal_dir.join(wal_file_name.clone() + ".partial");
-
-        {
-            let mut wal_file: File;
-            /* Try to open already completed segment */
-            if let Ok(file) = OpenOptions::new().write(true).open(&wal_file_path) {
-                wal_file = file;
-                partial = false;
-            } else if let Ok(file) = OpenOptions::new().write(true).open(&wal_file_partial_path) {
-                /* Try to open existed partial file */
-                wal_file = file;
-                partial = true;
-            } else {
-                /* Create and fill new partial file */
-                partial = true;
-                match OpenOptions::new()
-                    .create(true)
-                    .write(true)
-                    .open(&wal_file_partial_path)
-                {
-                    Ok(mut file) => {
-                        for _ in 0..(wal_seg_size / XLOG_BLCKSZ) {
-                            file.write_all(ZERO_BLOCK)?;
-                        }
-                        wal_file = file;
-                    }
-                    Err(e) => {
-                        error!("Failed to open log file {:?}: {}", &wal_file_path, e);
-                        return Err(e.into());
-                    }
-                }
-            }
-            wal_file.seek(SeekFrom::Start(xlogoff as u64))?;
-            wal_file.write_all(&buf[bytes_written..(bytes_written + bytes_to_write)])?;
-
-            // FIXME: Flush the file
-            //wal_file.sync_all()?;
-        }
-        /* Write was successful, advance our position */
-        bytes_written += bytes_to_write;
-        bytes_left -= bytes_to_write;
-        start_pos += bytes_to_write as u64;
-        xlogoff += bytes_to_write;
-
-        /* Did we reach the end of a WAL segment? */
-        if start_pos.segment_offset(wal_seg_size) == 0 {
-            xlogoff = 0;
-            if partial {
-                fs::rename(&wal_file_partial_path, &wal_file_path)?;
-            }
-        }
-    }
-    Ok(())
-}

pageserver/src/walredo.rs

@@ -298,9 +298,11 @@ impl PostgresRedoManager {
                     // Transaction manager stuff
                     let rec_segno = match rel {
                         RelishTag::Slru { slru, segno } => {
-                            if slru != SlruKind::Clog {
-                                panic!("Not valid XACT relish tag {:?}", rel);
-                            }
+                            assert!(
+                                slru == SlruKind::Clog,
+                                "Not valid XACT relish tag {:?}",
+                                rel
+                            );
                             segno
                         }
                         _ => panic!("Not valid XACT relish tag {:?}", rel),
@@ -420,7 +422,7 @@ impl PostgresRedoManager {
         );
 
         if let Err(e) = apply_result {
-            error!("could not apply WAL records: {}", e);
+            error!("could not apply WAL records: {:#}", e);
             result = Err(WalRedoError::IoError(e));
         } else {
             let img = apply_result.unwrap();
@@ -458,7 +460,7 @@ impl PostgresRedoProcess {
         if datadir.exists() {
             info!("directory {:?} exists, removing", &datadir);
             if let Err(e) = fs::remove_dir_all(&datadir) {
-                error!("could not remove old wal-redo-datadir: {:?}", e);
+                error!("could not remove old wal-redo-datadir: {:#}", e);
             }
         }
         info!("running initdb in {:?}", datadir.display());
@@ -544,8 +546,9 @@ impl PostgresRedoProcess {
         base_img: Option<Bytes>,
         records: &[WALRecord],
     ) -> Result<Bytes, std::io::Error> {
-        let stdin = &mut self.stdin;
         let stdout = &mut self.stdout;
+        // Buffer the writes to avoid a lot of small syscalls.
+        let mut stdin = tokio::io::BufWriter::new(&mut self.stdin);
 
         // We do three things simultaneously: send the old base image and WAL records to
         // the child process's stdin, read the result from child's stdout, and forward any logging

proxy/Cargo.toml

@@ -18,5 +18,6 @@ tokio = "1.11"
 tokio-postgres = { git = "https://github.com/zenithdb/rust-postgres.git", rev="9eb0dbfbeb6a6c1b79099b9f7ae4a8c021877858" }
 clap = "2.33.0"
 rustls = "0.19.1"
+reqwest = { version = "0.11", features = ["blocking", "json"] }
 
 zenith_utils = { path = "../zenith_utils" }

proxy/src/cplane_api.rs

@@ -1,17 +1,14 @@
-use anyhow::{bail, Result};
+use anyhow::{bail, Context, Result};
 use serde::{Deserialize, Serialize};
-use std::{
-    collections::HashMap,
-    net::{IpAddr, SocketAddr},
-};
+use std::net::{SocketAddr, ToSocketAddrs};
 
 pub struct CPlaneApi {
-    // address: SocketAddr,
+    auth_endpoint: &'static str,
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Debug)]
 pub struct DatabaseInfo {
-    pub host: IpAddr, // TODO: allow host name here too
+    pub host: String,
     pub port: u16,
     pub dbname: String,
     pub user: String,
@@ -19,8 +16,13 @@ pub struct DatabaseInfo {
 }
 
 impl DatabaseInfo {
-    pub fn socket_addr(&self) -> SocketAddr {
-        SocketAddr::new(self.host, self.port)
+    pub fn socket_addr(&self) -> Result<SocketAddr> {
+        let host_port = format!("{}:{}", self.host, self.port);
+        host_port
+            .to_socket_addrs()
+            .with_context(|| format!("cannot resolve {} to SocketAddr", host_port))?
+            .next()
+            .ok_or_else(|| anyhow::Error::msg("cannot resolve at least one SocketAddr"))
     }
 
     pub fn conn_string(&self) -> String {
@@ -31,62 +33,35 @@ impl DatabaseInfo {
     }
 }
 
-// mock cplane api
 impl CPlaneApi {
-    pub fn new(_address: &SocketAddr) -> CPlaneApi {
-        CPlaneApi {
-            // address: address.clone(),
-        }
+    pub fn new(auth_endpoint: &'static str) -> CPlaneApi {
+        CPlaneApi { auth_endpoint }
     }
 
-    pub fn check_auth(&self, user: &str, md5_response: &[u8], salt: &[u8; 4]) -> Result<()> {
-        // passwords for both is "mypass"
-        let auth_map: HashMap<_, &str> = vec![
-            ("stas@zenith", "716ee6e1c4a9364d66285452c47402b1"),
-            ("stas2@zenith", "3996f75df64c16a8bfaf01301b61d582"),
-        ]
-        .into_iter()
-        .collect();
+    pub fn authenticate_proxy_request(
+        &self,
+        user: &str,
+        database: &str,
+        md5_response: &[u8],
+        salt: &[u8; 4],
+    ) -> Result<DatabaseInfo> {
+        let mut url = reqwest::Url::parse(self.auth_endpoint)?;
+        url.query_pairs_mut()
+            .append_pair("login", user)
+            .append_pair("database", database)
+            .append_pair("md5response", std::str::from_utf8(md5_response)?)
+            .append_pair("salt", &hex::encode(salt));
 
-        let stored_hash = auth_map
-            .get(&user)
-            .ok_or_else(|| anyhow::Error::msg("user not found"))?;
-        let salted_stored_hash = format!(
-            "md5{:x}",
-            md5::compute([stored_hash.as_bytes(), salt].concat())
-        );
+        println!("cplane request: {}", url.as_str());
 
-        let received_hash = std::str::from_utf8(md5_response)?;
+        let resp = reqwest::blocking::get(url)?;
 
-        println!(
-            "auth: {} rh={} sh={} ssh={} {:?}",
-            user, received_hash, stored_hash, salted_stored_hash, salt
-        );
-
-        if received_hash == salted_stored_hash {
-            Ok(())
+        if resp.status().is_success() {
+            let conn_info: DatabaseInfo = serde_json::from_str(resp.text()?.as_str())?;
+            println!("got conn info: #{:?}", conn_info);
+            Ok(conn_info)
         } else {
             bail!("Auth failed")
         }
     }
-
-    pub fn get_database_uri(&self, _user: &str, _database: &str) -> Result<DatabaseInfo> {
-        Ok(DatabaseInfo {
-            host: "127.0.0.1".parse()?,
-            port: 5432,
-            dbname: "stas".to_string(),
-            user: "stas".to_string(),
-            password: "mypass".to_string(),
-        })
-    }
-
-    // pub fn create_database(&self, _user: &String, _database: &String) -> Result<DatabaseInfo> {
-    //     Ok(DatabaseInfo {
-    //         host: "127.0.0.1".parse()?,
-    //         port: 5432,
-    //         dbname: "stas".to_string(),
-    //         user: "stas".to_string(),
-    //         password: "mypass".to_string(),
-    //     })
-    // }
 }

proxy/src/main.rs

@@ -34,7 +34,7 @@ pub struct ProxyConf {
     pub redirect_uri: String,
 
     /// control plane address where we would check auth.
-    pub cplane_address: SocketAddr,
+    pub auth_endpoint: String,
 
     pub ssl_config: Option<Arc<ServerConfig>>,
 }
@@ -56,8 +56,7 @@ fn configure_ssl(arg_matches: &ArgMatches) -> anyhow::Result<Option<Arc<ServerCo
 
     let key = {
         let key_bytes = std::fs::read(key_path).context("SSL key file")?;
-        let mut keys = pemfile::rsa_private_keys(&mut &key_bytes[..])
-            .or_else(|_| pemfile::pkcs8_private_keys(&mut &key_bytes[..]))
+        let mut keys = pemfile::pkcs8_private_keys(&mut &key_bytes[..])
             .map_err(|_| anyhow!("couldn't read TLS keys"))?;
         ensure!(keys.len() == 1, "keys.len() = {} (should be 1)", keys.len());
         keys.pop().unwrap()
@@ -102,6 +101,14 @@ fn main() -> anyhow::Result<()> {
                 .help("redirect unauthenticated users to given uri")
                 .default_value("http://localhost:3000/psql_session/"),
         )
+        .arg(
+            Arg::with_name("auth-endpoint")
+                .short("a")
+                .long("auth-endpoint")
+                .takes_value(true)
+                .help("redirect unauthenticated users to given uri")
+                .default_value("http://localhost:3000/authenticate_proxy_request/"),
+        )
         .arg(
             Arg::with_name("ssl-key")
                 .short("k")
@@ -122,7 +129,7 @@ fn main() -> anyhow::Result<()> {
         proxy_address: arg_matches.value_of("proxy").unwrap().parse()?,
         mgmt_address: arg_matches.value_of("mgmt").unwrap().parse()?,
         redirect_uri: arg_matches.value_of("uri").unwrap().parse()?,
-        cplane_address: "127.0.0.1:3000".parse()?,
+        auth_endpoint: arg_matches.value_of("auth-endpoint").unwrap().parse()?,
         ssl_config: configure_ssl(&arg_matches)?,
     };
     let state = ProxyState {

proxy/src/mgmt.rs

@@ -49,7 +49,7 @@ struct MgmtHandler {
 //             "host": "127.0.0.1",
 //             "port": 5432,
 //             "dbname": "stas",
-//             "user": "stas"
+//             "user": "stas",
 //             "password": "mypass"
 //         }
 //     }
@@ -60,6 +60,9 @@ struct MgmtHandler {
 //         "Failure": "oops"
 //     }
 // }
+//
+// // to test manually by sending a query to mgmt interface:
+// psql -h 127.0.0.1 -p 9999 -c '{"session_id":"4f10dde522e14739","result":{"Success":{"host":"127.0.0.1","port":5432,"dbname":"stas","user":"stas","password":"stas"}}}'
 #[derive(Deserialize)]
 pub struct PsqlSessionResponse {
     session_id: String,
@@ -78,34 +81,47 @@ impl postgres_backend::Handler for MgmtHandler {
         pgb: &mut PostgresBackend,
         query_string: Bytes,
     ) -> anyhow::Result<()> {
-        let query_string = query_from_cstring(query_string);
+        let res = try_process_query(self, pgb, query_string);
+        // intercept and log error message
+        if res.is_err() {
+            println!("Mgmt query failed: #{:?}", res);
+        }
+        res
+    }
+}
 
-        println!("Got mgmt query: '{}'", std::str::from_utf8(&query_string)?);
+fn try_process_query(
+    mgmt: &mut MgmtHandler,
+    pgb: &mut PostgresBackend,
+    query_string: Bytes,
+) -> anyhow::Result<()> {
+    let query_string = query_from_cstring(query_string);
 
-        let resp: PsqlSessionResponse = serde_json::from_slice(&query_string)?;
+    println!("Got mgmt query: '{}'", std::str::from_utf8(&query_string)?);
 
-        let waiters = self.state.waiters.lock().unwrap();
+    let resp: PsqlSessionResponse = serde_json::from_slice(&query_string)?;
 
-        let sender = waiters
-            .get(&resp.session_id)
-            .ok_or_else(|| anyhow::Error::msg("psql_session_id is not found"))?;
+    let waiters = mgmt.state.waiters.lock().unwrap();
 
-        match resp.result {
-            PsqlSessionResult::Success(db_info) => {
-                sender.send(Ok(db_info))?;
+    let sender = waiters
+        .get(&resp.session_id)
+        .ok_or_else(|| anyhow::Error::msg("psql_session_id is not found"))?;
 
-                pgb.write_message_noflush(&SINGLE_COL_ROWDESC)?
-                    .write_message_noflush(&BeMessage::DataRow(&[Some(b"ok")]))?
-                    .write_message_noflush(&BeMessage::CommandComplete(b"SELECT 1"))?;
-                pgb.flush()?;
-                Ok(())
-            }
+    match resp.result {
+        PsqlSessionResult::Success(db_info) => {
+            sender.send(Ok(db_info))?;
 
-            PsqlSessionResult::Failure(message) => {
-                sender.send(Err(anyhow::Error::msg(message.clone())))?;
+            pgb.write_message_noflush(&SINGLE_COL_ROWDESC)?
+                .write_message_noflush(&BeMessage::DataRow(&[Some(b"ok")]))?
+                .write_message_noflush(&BeMessage::CommandComplete(b"SELECT 1"))?;
+            pgb.flush()?;
+            Ok(())
+        }
 
-                bail!("psql session request failed: {}", message)
-            }
+        PsqlSessionResult::Failure(message) => {
+            sender.send(Err(anyhow::Error::msg(message.clone())))?;
+
+            bail!("psql session request failed: {}", message)
         }
     }
 }

proxy/src/proxy.rs

@@ -57,7 +57,7 @@ pub fn proxy_conn_main(
 ) -> anyhow::Result<()> {
     let mut conn = ProxyConnection {
         state,
-        cplane: CPlaneApi::new(&state.conf.cplane_address),
+        cplane: CPlaneApi::new(&state.conf.auth_endpoint),
         user: "".into(),
         database: "".into(),
         pgb: PostgresBackend::new(
@@ -80,6 +80,8 @@ pub fn proxy_conn_main(
         conn.handle_new_user()?
     };
 
+    // XXX: move that inside handle_new_user/handle_existing_user to be able to
+    // report wrong connection error.
     proxy_pass(conn.pgb, db_info)
 }
 
@@ -172,21 +174,31 @@ impl ProxyConnection {
                 .split_last()
                 .ok_or_else(|| anyhow::Error::msg("unexpected password message"))?;
 
-            if let Err(e) = self.check_auth_md5(md5_response) {
-                self.pgb
-                    .write_message(&BeMessage::ErrorResponse(format!("{}", e)))?;
-                bail!("auth failed: {}", e);
-            } else {
-                self.pgb
-                    .write_message_noflush(&BeMessage::AuthenticationOk)?;
-                self.pgb
-                    .write_message_noflush(&BeMessage::ParameterStatus)?;
-                self.pgb.write_message(&BeMessage::ReadyForQuery)?;
-            }
-        }
+            match self.cplane.authenticate_proxy_request(
+                self.user.as_str(),
+                self.database.as_str(),
+                md5_response,
+                &self.md5_salt,
+            ) {
+                Err(e) => {
+                    self.pgb
+                        .write_message(&BeMessage::ErrorResponse(format!("{}", e)))?;
 
-        // ok, we are authorized
-        self.cplane.get_database_uri(&self.user, &self.database)
+                    bail!("auth failed: {}", e);
+                }
+                Ok(conn_info) => {
+                    self.pgb
+                        .write_message_noflush(&BeMessage::AuthenticationOk)?;
+                    self.pgb
+                        .write_message_noflush(&BeMessage::ParameterStatus)?;
+                    self.pgb.write_message(&BeMessage::ReadyForQuery)?;
+
+                    Ok(conn_info)
+                }
+            }
+        } else {
+            bail!("protocol violation");
+        }
     }
 
     fn handle_new_user(&mut self) -> anyhow::Result<DatabaseInfo> {
@@ -232,17 +244,11 @@ databases without opening the browser.
 
         Ok(dbinfo)
     }
-
-    fn check_auth_md5(&self, md5_response: &[u8]) -> anyhow::Result<()> {
-        assert!(self.is_existing_user());
-        self.cplane
-            .check_auth(self.user.as_str(), md5_response, &self.md5_salt)
-    }
 }
 
 /// Create a TCP connection to a postgres database, authenticate with it, and receive the ReadyForQuery message
 async fn connect_to_db(db_info: DatabaseInfo) -> anyhow::Result<tokio::net::TcpStream> {
-    let mut socket = tokio::net::TcpStream::connect(db_info.socket_addr()).await?;
+    let mut socket = tokio::net::TcpStream::connect(db_info.socket_addr()?).await?;
     let config = db_info.conn_string().parse::<tokio_postgres::Config>()?;
     let _ = config.connect_raw(&mut socket, NoTls).await?;
     Ok(socket)
@@ -273,7 +279,9 @@ fn proxy(
 
 /// Proxy a client connection to a postgres database
 fn proxy_pass(pgb: PostgresBackend, db_info: DatabaseInfo) -> anyhow::Result<()> {
-    let runtime = tokio::runtime::Builder::new_current_thread().build()?;
+    let runtime = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()?;
     let db_stream = runtime.block_on(connect_to_db(db_info))?;
     let db_stream = db_stream.into_std()?;
     db_stream.set_nonblocking(false)?;

test_runner/batch_others/test_restart_compute.py

@@ -9,10 +9,7 @@ pytest_plugins = ("fixtures.zenith_fixtures")
 #
 # Test restarting and recreating a postgres instance
 #
-# XXX: with_wal_acceptors=True fails now, would be fixed with
-# `postgres --sync-walkeepers` patches.
-#
-@pytest.mark.parametrize('with_wal_acceptors', [False])
+@pytest.mark.parametrize('with_wal_acceptors', [False, True])
 def test_restart_compute(
         zenith_cli,
         pageserver: ZenithPageserver,

test_runner/batch_others/test_wal_acceptor.py

@@ -1,10 +1,14 @@
 import pytest
 import random
 import time
+import os
+import subprocess
+import uuid
 
 from contextlib import closing
 from multiprocessing import Process, Value
-from fixtures.zenith_fixtures import WalAcceptorFactory, ZenithPageserver, PostgresFactory
+from fixtures.zenith_fixtures import WalAcceptorFactory, ZenithPageserver, PostgresFactory, PgBin
+from fixtures.utils import lsn_to_hex, mkdir_if_needed
 
 pytest_plugins = ("fixtures.zenith_fixtures")
 
@@ -198,3 +202,92 @@ def test_race_conditions(zenith_cli, pageserver: ZenithPageserver, postgres: Pos
 
     stop_value.value = 1
     proc.join()
+
+class ProposerPostgres:
+    """Object for running safekeepers sync with walproposer"""
+    def __init__(self, pgdata_dir: str, pg_bin: PgBin, timeline_id: str, tenant_id: str):
+        self.pgdata_dir: str = pgdata_dir
+        self.pg_bin: PgBin = pg_bin
+        self.timeline_id: str = timeline_id
+        self.tenant_id: str = tenant_id
+
+    def pg_data_dir_path(self) -> str:
+        """ Path to data directory """
+        return self.pgdata_dir
+
+    def config_file_path(self) -> str:
+        """ Path to postgresql.conf """
+        return os.path.join(self.pgdata_dir, 'postgresql.conf')
+
+    def create_dir_config(self, wal_acceptors: str):
+        """ Create dir and config for running --sync-safekeepers """
+
+        mkdir_if_needed(self.pg_data_dir_path())
+        with open(self.config_file_path(), "w") as f:
+            f.writelines([
+                "synchronous_standby_names = 'walproposer'\n",
+                f"zenith.zenith_timeline = '{self.timeline_id}'\n",
+                f"zenith.zenith_tenant = '{self.tenant_id}'\n",
+                f"wal_acceptors = '{wal_acceptors}'\n",
+            ])
+
+    def sync_safekeepers(self) -> str:
+        """
+        Run 'postgres --sync-safekeepers'.
+        Returns execution result, which is commit_lsn after sync.
+        """
+
+        command = ["postgres", "--sync-safekeepers"]
+        env = {
+            "PGDATA": self.pg_data_dir_path(),
+        }
+
+        basepath = self.pg_bin.run_capture(command, env)
+        stdout_filename = basepath + '.stdout'
+
+        with open(stdout_filename, 'r') as stdout_f:
+            stdout = stdout_f.read()
+            return stdout.strip("\n ")
+
+
+# insert wal in all safekeepers and run sync on proposer
+def test_sync_safekeepers(repo_dir: str, pg_bin: PgBin, wa_factory: WalAcceptorFactory):
+    wa_factory.start_n_new(3)
+
+    timeline_id = uuid.uuid4().hex
+    tenant_id = uuid.uuid4().hex
+
+    # write config for proposer
+    pgdata_dir = os.path.join(repo_dir, "proposer_pgdata")
+    pg = ProposerPostgres(pgdata_dir, pg_bin, timeline_id, tenant_id)
+    pg.create_dir_config(wa_factory.get_connstrs())
+
+    # valid lsn, which is not in the segment start, nor in zero segment
+    epoch_start_lsn = 0x16B9188  # 0/16B9188
+    begin_lsn = epoch_start_lsn
+
+    # append and commit WAL
+    lsn_after_append = []
+    for i in range(3):
+        res = wa_factory.instances[i].append_logical_message(
+            tenant_id,
+            timeline_id,
+            {
+                "lm_prefix": "prefix",
+                "lm_message": "message",
+                "set_commit_lsn": True,
+                "term": 2,
+                "begin_lsn": begin_lsn,
+                "epoch_start_lsn": epoch_start_lsn,
+                "truncate_lsn": epoch_start_lsn,
+            },
+        )
+        lsn_hex = lsn_to_hex(res["inserted_wal"]["end_lsn"])
+        lsn_after_append.append(lsn_hex)
+        print(f"safekeeper[{i}] lsn after append: {lsn_hex}")
+
+    # run sync safekeepers
+    lsn_after_sync = pg.sync_safekeepers()
+    print(f"lsn after sync = {lsn_after_sync}")
+
+    assert all(lsn_after_sync == lsn for lsn in lsn_after_append)

test_runner/batch_others/test_wal_acceptor_async.py

@@ -118,9 +118,14 @@ async def run_restarts_under_load(pg: Postgres, acceptors: List[WalAcceptor], n_
         victim = acceptors[it % len(acceptors)]
         victim.stop()
 
-        # wait for transactions that could have started and finished before
-        # victim acceptor was stopped
-        await asyncio.sleep(1)
+        # Wait till previous victim recovers so it is ready for the next
+        # iteration by making any writing xact.
+        conn = await pg.connect_async()
+        await conn.execute(
+            'UPDATE bank_accs SET amount = amount WHERE uid = 1',
+            timeout=120
+        )
+        await conn.close()
 
         stats.reset()
         await asyncio.sleep(period_time)

test_runner/batch_others/test_wal_proposer.py

@@ -1,103 +0,0 @@
-import os
-import subprocess
-import uuid
-
-from fixtures.zenith_fixtures import WalAcceptorFactory, PgBin
-from fixtures.utils import lsn_to_hex, mkdir_if_needed
-
-pytest_plugins = ("fixtures.zenith_fixtures")
-
-
-class ProposerPostgres:
-    """Object for running safekeepers sync with walproposer"""
-    def __init__(self, pgdata_dir: str, pg_bin: PgBin, timeline_id: str, tenant_id: str):
-        self.pgdata_dir: str = pgdata_dir
-        self.pg_bin: PgBin = pg_bin
-        self.timeline_id: str = timeline_id
-        self.tenant_id: str = tenant_id
-
-    def pg_data_dir_path(self) -> str:
-        """ Path to data directory """
-        return self.pgdata_dir
-
-    def config_file_path(self) -> str:
-        """ Path to postgresql.conf """
-        return os.path.join(self.pgdata_dir, 'postgresql.conf')
-
-    def create_dir_config(self, wal_acceptors: str):
-        """ Create dir and config for running --sync-safekeepers """
-
-        mkdir_if_needed(self.pg_data_dir_path())
-        with open(self.config_file_path(), "w") as f:
-            f.write("zenith.zenith_timeline = '{}'\n".format(self.timeline_id))
-            f.write("zenith.zenith_tenant = '{}'\n".format(self.tenant_id))
-            f.write("synchronous_standby_names = '{}'\n".format("walproposer"))
-            f.write("wal_acceptors = '{}'\n".format(wal_acceptors))
-
-    def sync_safekeepers(self) -> subprocess.CompletedProcess:
-        """
-        Run 'postgres --sync-safekeepers'.
-        Returns execution result, which is commit_lsn after sync.
-        """
-
-        pg_path = os.path.join(self.pg_bin.pg_bin_path, "postgres")
-        command = [pg_path, "--sync-safekeepers"]
-        env = {
-            "PGDATA": self.pg_data_dir_path(),
-        }
-
-        print('Running command "{}"'.format(" ".join(command)))
-        res = subprocess.run(
-            command, env=env, check=True, text=True, stdout=subprocess.PIPE
-        )
-
-        return res.stdout.strip("\n ")
-
-
-# insert wal in all safekeepers and run sync on proposer
-def test_sync_safekeepers(repo_dir: str, pg_bin: PgBin, wa_factory: WalAcceptorFactory):
-    wa_factory.start_n_new(3)
-
-    timeline_id = uuid.uuid4().hex
-    tenant_id = uuid.uuid4().hex
-
-    # write config for proposer
-    pgdata_dir = os.path.join(repo_dir, "proposer_pgdata")
-    pg = ProposerPostgres(pgdata_dir, pg_bin, timeline_id, tenant_id)
-    pg.create_dir_config(wa_factory.get_connstrs())
-
-    # run sync to init safekeepers with ProposerGreeting
-    initial_lsn = pg.sync_safekeepers()
-
-    # should be 0/0 for empty safekeepers
-    assert initial_lsn == "0/0"
-
-    # valid lsn, which is not in the segment start, nor in zero segment
-    epoch_start_lsn = 0x16B9188  # 0/16B9188
-    begin_lsn = epoch_start_lsn
-
-    # append and commit WAL
-    lsn_after_append = []
-    for i in range(3):
-        res = wa_factory.instances[i].append_logical_message(
-            tenant_id,
-            timeline_id,
-            {
-                "lm_prefix": "prefix",
-                "lm_message": "message",
-                "set_commit_lsn": True,
-                "term": 2,
-                "begin_lsn": begin_lsn,
-                "epoch_start_lsn": epoch_start_lsn,
-                "truncate_lsn": epoch_start_lsn,
-            },
-        )
-        lsn_hex = lsn_to_hex(res["inserted_wal"]["end_lsn"])
-        lsn_after_append.append(lsn_hex)
-        print(f"safekeeper[{i}] lsn after append: {lsn_hex}")
-
-    # run sync safekeepers
-    lsn_after_sync = pg.sync_safekeepers()
-    print(f"lsn after sync = {lsn_after_sync}")
-
-    assert all(lsn_after_sync == lsn for lsn in lsn_after_append)

test_runner/fixtures/utils.py

@@ -21,7 +21,7 @@ def mkdir_if_needed(path: str) -> None:
     assert os.path.isdir(path)
 
 
-def subprocess_capture(capture_dir: str, cmd: List[str], **kwargs: Any) -> None:
+def subprocess_capture(capture_dir: str, cmd: List[str], **kwargs: Any) -> str:
     """ Run a process and capture its output
 
     Output will go to files named "cmd_NNN.stdout" and "cmd_NNN.stderr"
@@ -29,6 +29,7 @@ def subprocess_capture(capture_dir: str, cmd: List[str], **kwargs: Any) -> None:
     counter.
 
     If those files already exist, we will overwrite them.
+    Returns basepath for files with captured output.
     """
     assert type(cmd) is list
     base = os.path.basename(cmd[0]) + '_{}'.format(global_counter())
@@ -41,6 +42,8 @@ def subprocess_capture(capture_dir: str, cmd: List[str], **kwargs: Any) -> None:
             print('(capturing output to "{}.stdout")'.format(base))
             subprocess.run(cmd, **kwargs, stdout=stdout_f, stderr=stderr_f)
 
+    return basepath
+
 
 _global_counter = 0
 

test_runner/fixtures/zenith_fixtures.py

@@ -486,17 +486,19 @@ class PgBin:
     def run_capture(self,
                     command: List[str],
                     env: Optional[Env] = None,
-                    cwd: Optional[str] = None) -> None:
+                    cwd: Optional[str] = None,
+                    **kwargs: Any) -> None:
         """
         Run one of the postgres binaries, with stderr and stdout redirected to a file.
 
-        This is just like `run`, but for chatty programs.
+        This is just like `run`, but for chatty programs. Returns basepath for files
+        with captured output.
         """
 
         self._fixpath(command)
         print('Running command "{}"'.format(' '.join(command)))
         env = self._build_env(env)
-        subprocess_capture(self.log_dir, command, env=env, cwd=cwd, check=True)
+        return subprocess_capture(self.log_dir, command, env=env, cwd=cwd, check=True, **kwargs)
 
 
 @zenfixture

test_runner/performance/test_bulk_tenant_create.py

@@ -0,0 +1,58 @@
+import timeit
+import pytest
+
+from fixtures.zenith_fixtures import (
+    TenantFactory,
+    ZenithCli,
+    PostgresFactory,
+)
+
+pytest_plugins = ("fixtures.benchmark_fixture")
+
+# Run bulk tenant creation test.
+#
+# Collects metrics:
+#
+# 1. Time to create {1,10,50} tenants
+# 2. Average creation time per tenant
+
+
+@pytest.mark.parametrize('tenants_count', [1, 5, 10])
+@pytest.mark.parametrize('use_wal_acceptors', ['with_wa', 'without_wa'])
+def test_bulk_tenant_create(
+    zenith_cli: ZenithCli,
+    tenant_factory: TenantFactory,
+    postgres: PostgresFactory,
+    wa_factory,
+    use_wal_acceptors: str,
+    tenants_count: int,
+    zenbenchmark,
+):
+    """Measure tenant creation time (with and without wal acceptors)"""
+
+    time_slices = []
+
+    for i in range(tenants_count):
+        start = timeit.default_timer()
+
+        tenant = tenant_factory.create()
+        zenith_cli.run([
+            "branch", f"test_bulk_tenant_create_{tenants_count}_{i}_{use_wal_acceptors}", "main",
+            f"--tenantid={tenant}"
+        ])
+
+        if use_wal_acceptors == 'with_wa':
+            wa_factory.start_n_new(3)
+
+        pg_tenant = postgres.create_start(
+            f"test_bulk_tenant_create_{tenants_count}_{i}_{use_wal_acceptors}",
+            tenant,
+            wal_acceptors=wa_factory.get_connstrs() if use_wal_acceptors == 'with_wa' else None,
+        )
+
+        end = timeit.default_timer()
+        time_slices.append(end - start)
+
+        pg_tenant.stop()
+
+    zenbenchmark.record('tenant_creation_time', sum(time_slices) / len(time_slices), 's')

walkeeper/src/json_ctrl.rs

@@ -13,7 +13,9 @@ use log::*;
 use serde::{Deserialize, Serialize};
 
 use crate::safekeeper::{AcceptorProposerMessage, AppendResponse};
-use crate::safekeeper::{AppendRequest, AppendRequestHeader, ProposerAcceptorMessage};
+use crate::safekeeper::{
+    AppendRequest, AppendRequestHeader, ProposerAcceptorMessage, ProposerGreeting,
+};
 use crate::safekeeper::{SafeKeeperState, Term};
 use crate::send_wal::SendWalHandler;
 use crate::timeline::TimelineTools;
@@ -48,6 +50,9 @@ struct AppendResult {
     inserted_wal: InsertedWAL,
 }
 
+/// Handles command to craft logical message WAL record with given
+/// content, and then append it with specified term and lsn. This
+/// function is used to test safekeepers in different scenarios.
 pub fn handle_json_ctrl(
     swh: &mut SendWalHandler,
     pgb: &mut PostgresBackend,
@@ -62,6 +67,9 @@ pub fn handle_json_ctrl(
     let append_request: AppendLogicalMessage = serde_json::from_slice(cmd)?;
     info!("JSON_CTRL request: {:?}", append_request);
 
+    // need to init safekeeper state before AppendRequest
+    prepare_safekeeper(swh)?;
+
     let inserted_wal = append_logical_message(swh, append_request)?;
     let response = AppendResult {
         state: swh.timeline.get().get_info(),
@@ -80,6 +88,27 @@ pub fn handle_json_ctrl(
     Ok(())
 }
 
+/// Prepare safekeeper to process append requests without crashes,
+/// by sending ProposerGreeting with default server.wal_seg_size.
+fn prepare_safekeeper(swh: &mut SendWalHandler) -> Result<()> {
+    let greeting_request = ProposerAcceptorMessage::Greeting(ProposerGreeting {
+        protocol_version: 1, // current protocol
+        pg_version: 0,       // unknown
+        proposer_id: [0u8; 16],
+        system_id: 0,
+        ztli: swh.timelineid.unwrap(),
+        tenant_id: swh.tenantid.unwrap(),
+        tli: 0,
+        wal_seg_size: pg_constants::WAL_SEGMENT_SIZE as u32, // 16MB, default for tests
+    });
+
+    let response = swh.timeline.get().process_msg(&greeting_request)?;
+    match response {
+        AcceptorProposerMessage::Greeting(_) => Ok(()),
+        _ => anyhow::bail!("not GreetingResponse"),
+    }
+}
+
 #[derive(Serialize, Deserialize)]
 struct InsertedWAL {
     begin_lsn: Lsn,
@@ -122,7 +151,7 @@ fn append_logical_message(
 
     let append_response = match response {
         AcceptorProposerMessage::AppendResponse(resp) => resp,
-        _ => return Err(anyhow!("not AppendResponse")),
+        _ => anyhow::bail!("not AppendResponse"),
     };
 
     Ok(InsertedWAL {
@@ -148,9 +177,9 @@ impl XlLogicalMessage {
     }
 }
 
-// Create new WAL record for non-transactional logical message.
-// Used for creating artificial WAL for tests, as LogicalMessage
-// record is basically no-op.
+/// Create new WAL record for non-transactional logical message.
+/// Used for creating artificial WAL for tests, as LogicalMessage
+/// record is basically no-op.
 fn encode_logical_message(prefix: String, message: String) -> Vec<u8> {
     let mut prefix_bytes = BytesMut::with_capacity(prefix.len() + 1);
     prefix_bytes.put(prefix.as_bytes());

walkeeper/src/safekeeper.rs

@@ -275,8 +275,8 @@ impl AcceptorProposerMessage {
 pub trait Storage {
     /// Persist safekeeper state on disk, optionally syncing it.
     fn persist(&mut self, s: &SafeKeeperState, sync: bool) -> Result<()>;
-    /// Write piece of wal in buf to disk.
-    fn write_wal(&mut self, s: &SafeKeeperState, startpos: Lsn, buf: &[u8]) -> Result<()>;
+    /// Write piece of wal in buf to disk and sync it.
+    fn write_wal(&mut self, server: &ServerInfo, startpos: Lsn, buf: &[u8]) -> Result<()>;
 }
 
 /// SafeKeeper which consumes events (messages from compute) and provides
@@ -423,7 +423,7 @@ where
         let mut last_rec_lsn = Lsn(0);
         if !msg.wal_data.is_empty() {
             self.storage
-                .write_wal(&self.s, msg.h.begin_lsn, &msg.wal_data)?;
+                .write_wal(&self.s.server, msg.h.begin_lsn, &msg.wal_data)?;
 
             // figure out last record's end lsn for reporting (if we got the
             // whole record)
@@ -546,7 +546,7 @@ mod tests {
             Ok(())
         }
 
-        fn write_wal(&mut self, _s: &SafeKeeperState, _startpos: Lsn, _buf: &[u8]) -> Result<()> {
+        fn write_wal(&mut self, _server: &ServerInfo, _startpos: Lsn, _buf: &[u8]) -> Result<()> {
             Ok(())
         }
     }

walkeeper/src/send_wal.rs

@@ -50,9 +50,11 @@ impl postgres_backend::Handler for SendWalHandler {
     }
 
     fn process_query(&mut self, pgb: &mut PostgresBackend, query_string: Bytes) -> Result<()> {
-        // START_WAL_PUSH is the only command that initializes the timeline
+        // START_WAL_PUSH is the only command that initializes the timeline in production.
+        // There is also JSON_CTRL command, which should initialize the timeline for testing.
         if self.timeline.is_none() {
-            if query_string.starts_with(b"START_WAL_PUSH") {
+            if query_string.starts_with(b"START_WAL_PUSH") || query_string.starts_with(b"JSON_CTRL")
+            {
                 self.timeline.set(
                     &self.conf,
                     self.tenantid.unwrap(),

walkeeper/src/timeline.rs

@@ -18,8 +18,8 @@ use zenith_utils::zid::{ZTenantId, ZTimelineId};
 
 use crate::replication::{HotStandbyFeedback, END_REPLICATION_MARKER};
 use crate::safekeeper::{
-    AcceptorProposerMessage, ProposerAcceptorMessage, SafeKeeper, SafeKeeperState, Storage,
-    SK_FORMAT_VERSION, SK_MAGIC,
+    AcceptorProposerMessage, ProposerAcceptorMessage, SafeKeeper, SafeKeeperState, ServerInfo,
+    Storage, SK_FORMAT_VERSION, SK_MAGIC,
 };
 use crate::WalAcceptorConf;
 use postgres_ffi::xlog_utils::{XLogFileName, XLOG_BLCKSZ};
@@ -32,7 +32,7 @@ struct SharedState {
     sk: SafeKeeper<FileStorage>,
     /// For receiving-sending wal cooperation
     /// quorum commit LSN we've notified walsenders about
-    commit_lsn: Lsn,
+    notified_commit_lsn: Lsn,
     /// combined hot standby feedback from all replicas
     hs_feedback: HotStandbyFeedback,
 }
@@ -72,7 +72,7 @@ impl SharedState {
         };
 
         Ok(Self {
-            commit_lsn: Lsn(0),
+            notified_commit_lsn: Lsn(0),
             sk: SafeKeeper::new(Lsn(flush_lsn), tli, storage, state),
             hs_feedback: HotStandbyFeedback {
                 ts: 0,
@@ -186,7 +186,7 @@ impl Timeline {
     pub fn wait_for_lsn(&self, lsn: Lsn) -> Lsn {
         let mut shared_state = self.mutex.lock().unwrap();
         loop {
-            let commit_lsn = shared_state.commit_lsn;
+            let commit_lsn = shared_state.notified_commit_lsn;
             // This must be `>`, not `>=`.
             if commit_lsn > lsn {
                 return commit_lsn;
@@ -198,8 +198,8 @@ impl Timeline {
     // Notify caught-up WAL senders about new WAL data received
     pub fn notify_wal_senders(&self, commit_lsn: Lsn) {
         let mut shared_state = self.mutex.lock().unwrap();
-        if shared_state.commit_lsn < commit_lsn {
-            shared_state.commit_lsn = commit_lsn;
+        if shared_state.notified_commit_lsn < commit_lsn {
+            shared_state.notified_commit_lsn = commit_lsn;
             self.cond.notify_all();
         }
     }
@@ -337,14 +337,14 @@ impl Storage for FileStorage {
         Ok(())
     }
 
-    fn write_wal(&mut self, s: &SafeKeeperState, startpos: Lsn, buf: &[u8]) -> Result<()> {
+    fn write_wal(&mut self, server: &ServerInfo, startpos: Lsn, buf: &[u8]) -> Result<()> {
         let mut bytes_left: usize = buf.len();
         let mut bytes_written: usize = 0;
         let mut partial;
         let mut start_pos = startpos;
         const ZERO_BLOCK: &[u8] = &[0u8; XLOG_BLCKSZ];
-        let wal_seg_size = s.server.wal_seg_size as usize;
-        let ztli = s.server.ztli;
+        let wal_seg_size = server.wal_seg_size as usize;
+        let ztli = server.ztli;
 
         /* Extract WAL location for this block */
         let mut xlogoff = start_pos.segment_offset(wal_seg_size) as usize;
@@ -365,7 +365,7 @@ impl Storage for FileStorage {
             /* Open file */
             let segno = start_pos.segment_number(wal_seg_size);
             // note: we basically don't support changing pg timeline
-            let wal_file_name = XLogFileName(s.server.tli, segno, wal_seg_size);
+            let wal_file_name = XLogFileName(server.tli, segno, wal_seg_size);
             let wal_file_path = self
                 .conf
                 .data_dir

zenith_utils/Cargo.toml

@@ -38,3 +38,4 @@ rustls-split = "0.2.1"
 hex-literal = "0.3"
 bytes = "1.0"
 webpki = "0.21"
+tempfile = "3.2"

zenith_utils/src/crashsafe_dir.rs

@@ -0,0 +1,125 @@
+use std::{
+    fs::{self, File},
+    io,
+    path::Path,
+};
+
+/// Similar to [`std::fs::create_dir`], except we fsync the
+/// created directory and its parent.
+pub fn create_dir(path: impl AsRef<Path>) -> io::Result<()> {
+    let path = path.as_ref();
+
+    fs::create_dir(path)?;
+    File::open(path)?.sync_all()?;
+
+    if let Some(parent) = path.parent() {
+        File::open(parent)?.sync_all()
+    } else {
+        Err(io::Error::new(
+            io::ErrorKind::InvalidInput,
+            "can't find parent",
+        ))
+    }
+}
+
+/// Similar to [`std::fs::create_dir_all`], except we fsync all
+/// newly created directories and the pre-existing parent.
+pub fn create_dir_all(path: impl AsRef<Path>) -> io::Result<()> {
+    let mut path = path.as_ref();
+
+    let mut dirs_to_create = Vec::new();
+
+    // Figure out which directories we need to create.
+    loop {
+        match path.metadata() {
+            Ok(metadata) if metadata.is_dir() => break,
+            Ok(_) => {
+                return Err(io::Error::new(
+                    io::ErrorKind::AlreadyExists,
+                    format!("non-directory found in path: {:?}", path),
+                ));
+            }
+            Err(ref e) if e.kind() == io::ErrorKind::NotFound => {}
+            Err(e) => return Err(e),
+        }
+
+        dirs_to_create.push(path);
+
+        match path.parent() {
+            Some(parent) => path = parent,
+            None => {
+                return Err(io::Error::new(
+                    io::ErrorKind::InvalidInput,
+                    "can't find parent",
+                ))
+            }
+        }
+    }
+
+    // Create directories from parent to child.
+    for &path in dirs_to_create.iter().rev() {
+        fs::create_dir(path)?;
+    }
+
+    // Fsync the created directories from child to parent.
+    for &path in dirs_to_create.iter() {
+        File::open(path)?.sync_all()?;
+    }
+
+    // If we created any new directories, fsync the parent.
+    if !dirs_to_create.is_empty() {
+        File::open(path)?.sync_all()?;
+    }
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use tempfile::tempdir;
+
+    use super::*;
+
+    #[test]
+    fn test_create_dir_fsyncd() {
+        let dir = tempdir().unwrap();
+
+        let existing_dir_path = dir.path();
+        let err = create_dir(existing_dir_path).unwrap_err();
+        assert_eq!(err.kind(), io::ErrorKind::AlreadyExists);
+
+        let child_dir = existing_dir_path.join("child");
+        create_dir(child_dir).unwrap();
+
+        let nested_child_dir = existing_dir_path.join("child1").join("child2");
+        let err = create_dir(nested_child_dir).unwrap_err();
+        assert_eq!(err.kind(), io::ErrorKind::NotFound);
+    }
+
+    #[test]
+    fn test_create_dir_all_fsyncd() {
+        let dir = tempdir().unwrap();
+
+        let existing_dir_path = dir.path();
+        create_dir_all(existing_dir_path).unwrap();
+
+        let child_dir = existing_dir_path.join("child");
+        assert!(!child_dir.exists());
+        create_dir_all(&child_dir).unwrap();
+        assert!(child_dir.exists());
+
+        let nested_child_dir = existing_dir_path.join("child1").join("child2");
+        assert!(!nested_child_dir.exists());
+        create_dir_all(&nested_child_dir).unwrap();
+        assert!(nested_child_dir.exists());
+
+        let file_path = existing_dir_path.join("file");
+        std::fs::write(&file_path, b"").unwrap();
+
+        let err = create_dir_all(&file_path).unwrap_err();
+        assert_eq!(err.kind(), io::ErrorKind::AlreadyExists);
+
+        let invalid_dir_path = file_path.join("folder");
+        create_dir_all(&invalid_dir_path).unwrap_err();
+    }
+}

zenith_utils/src/lib.rs

@@ -18,6 +18,9 @@ pub mod pq_proto;
 // dealing with connstring parsing and handy access to it's parts
 pub mod connstring;
 
+// helper functions for creating and fsyncing directories/trees
+pub mod crashsafe_dir;
+
 // common authentication routines
 pub mod auth;
 

zenith_utils/src/lsn.rs

@@ -192,9 +192,7 @@ impl AtomicLsn {
     /// This operation will panic on overflow.
     pub fn fetch_add(&self, val: u64) -> Lsn {
         let prev = self.inner.fetch_add(val, Ordering::AcqRel);
-        if prev.checked_add(val).is_none() {
-            panic!("AtomicLsn overflow");
-        }
+        assert!(prev.checked_add(val).is_some(), "AtomicLsn overflow");
         Lsn(prev)
     }
 

zenith_utils/src/pq_proto.rs

@@ -652,7 +652,7 @@ impl<'a> BeMessage<'a> {
             }
 
             BeMessage::EncryptionResponse(should_negotiate) => {
-                let response = if *should_negotiate { b'Y' } else { b'N' };
+                let response = if *should_negotiate { b'S' } else { b'N' };
                 buf.put_u8(response);
             }
 

zenith_utils/src/zid.rs

@@ -78,6 +78,10 @@ macro_rules! zid_newtype {
             pub fn generate() -> Self {
                 $t(ZId::generate())
             }
+
+            pub const fn from_array(b: [u8; 16]) -> Self {
+                $t(ZId(b))
+            }
         }
 
         impl FromStr for $t {

zenith_utils/tests/ssl_test.rs

@@ -43,7 +43,7 @@ fn ssl() {
         client_sock.write_u32::<BigEndian>(80877103).unwrap();
 
         let ssl_response = client_sock.read_u8().unwrap();
-        assert_eq!(b'Y', ssl_response);
+        assert_eq!(b'S', ssl_response);
 
         let mut cfg = rustls::ClientConfig::new();
         cfg.root_store.add(&CERT).unwrap();