pageserver - call fsync in large batches

Doesn't parallelize the calls yet, just accumulates a number of files
before fsyncing.

Cargo.lock

@@ -1213,6 +1213,7 @@ dependencies = [
  "tokio",
  "toml",
  "tracing",
+ "url",
  "workspace_hack",
  "zenith_metrics",
  "zenith_utils",
@@ -2577,6 +2578,7 @@ dependencies = [
  "hyper",
  "jsonwebtoken",
  "lazy_static",
+ "nix",
  "postgres",
  "rand",
  "routerify",

pageserver/Cargo.toml

@@ -37,6 +37,7 @@ async-trait = "0.1"
 const_format = "0.2.21"
 tracing = "0.1.27"
 signal-hook = {version = "0.3.10", features = ["extended-siginfo"] }
+url = "2"
 
 postgres_ffi = { path = "../postgres_ffi" }
 zenith_metrics = { path = "../zenith_metrics" }

pageserver/src/bin/pageserver.rs

@@ -5,13 +5,12 @@
 use serde::{Deserialize, Serialize};
 use std::{
     env,
-    net::TcpListener,
     path::{Path, PathBuf},
     str::FromStr,
     thread,
 };
 use tracing::*;
-use zenith_utils::{auth::JwtAuth, logging, postgres_backend::AuthType};
+use zenith_utils::{auth::JwtAuth, logging, postgres_backend::AuthType, tcp_listener};
 
 use anyhow::{bail, ensure, Context, Result};
 use signal_hook::consts::signal::*;
@@ -44,6 +43,7 @@ struct CfgFileParams {
     checkpoint_period: Option<String>,
     gc_horizon: Option<String>,
     gc_period: Option<String>,
+    open_mem_limit: Option<String>,
     pg_distrib_dir: Option<String>,
     auth_validation_public_key_path: Option<String>,
     auth_type: Option<String>,
@@ -105,6 +105,7 @@ impl CfgFileParams {
             checkpoint_period: get_arg("checkpoint_period"),
             gc_horizon: get_arg("gc_horizon"),
             gc_period: get_arg("gc_period"),
+            open_mem_limit: get_arg("open_mem_limit"),
             pg_distrib_dir: get_arg("postgres-distrib"),
             auth_validation_public_key_path: get_arg("auth-validation-public-key-path"),
             auth_type: get_arg("auth-type"),
@@ -123,6 +124,7 @@ impl CfgFileParams {
             checkpoint_period: self.checkpoint_period.or(other.checkpoint_period),
             gc_horizon: self.gc_horizon.or(other.gc_horizon),
             gc_period: self.gc_period.or(other.gc_period),
+            open_mem_limit: self.open_mem_limit.or(other.open_mem_limit),
             pg_distrib_dir: self.pg_distrib_dir.or(other.pg_distrib_dir),
             auth_validation_public_key_path: self
                 .auth_validation_public_key_path
@@ -167,6 +169,11 @@ impl CfgFileParams {
             None => DEFAULT_GC_PERIOD,
         };
 
+        let open_mem_limit: usize = match self.open_mem_limit.as_ref() {
+            Some(open_mem_limit_str) => open_mem_limit_str.parse()?,
+            None => DEFAULT_OPEN_MEM_LIMIT,
+        };
+
         let pg_distrib_dir = match self.pg_distrib_dir.as_ref() {
             Some(pg_distrib_dir_str) => PathBuf::from(pg_distrib_dir_str),
             None => env::current_dir()?.join("tmp_install"),
@@ -238,6 +245,7 @@ impl CfgFileParams {
             checkpoint_period,
             gc_horizon,
             gc_period,
+            open_mem_limit,
 
             superuser: String::from(DEFAULT_SUPERUSER),
 
@@ -308,6 +316,12 @@ fn main() -> Result<()> {
                 .takes_value(true)
                 .help("Interval between garbage collector iterations"),
         )
+        .arg(
+            Arg::with_name("open_mem_limit")
+                .long("open_mem_limit")
+                .takes_value(true)
+                .help("Amount of memory reserved for buffering incoming WAL"),
+        )
         .arg(
             Arg::with_name("workdir")
                 .short("D")
@@ -480,13 +494,13 @@ fn start_pageserver(conf: &'static PageServerConf) -> Result<()> {
         "Starting pageserver http handler on {}",
         conf.listen_http_addr
     );
-    let http_listener = TcpListener::bind(conf.listen_http_addr.clone())?;
+    let http_listener = tcp_listener::bind(conf.listen_http_addr.clone())?;
 
     info!(
         "Starting pageserver pg protocol handler on {}",
         conf.listen_pg_addr
     );
-    let pageserver_listener = TcpListener::bind(conf.listen_pg_addr.clone())?;
+    let pageserver_listener = tcp_listener::bind(conf.listen_pg_addr.clone())?;
 
     if conf.daemonize {
         info!("daemonizing...");
@@ -602,6 +616,7 @@ mod tests {
             checkpoint_period: Some("checkpoint_period_VALUE".to_string()),
             gc_horizon: Some("gc_horizon_VALUE".to_string()),
             gc_period: Some("gc_period_VALUE".to_string()),
+            open_mem_limit: Some("open_mem_limit_VALUE".to_string()),
             pg_distrib_dir: Some("pg_distrib_dir_VALUE".to_string()),
             auth_validation_public_key_path: Some(
                 "auth_validation_public_key_path_VALUE".to_string(),
@@ -625,6 +640,7 @@ checkpoint_distance = 'checkpoint_distance_VALUE'
 checkpoint_period = 'checkpoint_period_VALUE'
 gc_horizon = 'gc_horizon_VALUE'
 gc_period = 'gc_period_VALUE'
+open_mem_limit = 'open_mem_limit_VALUE'
 pg_distrib_dir = 'pg_distrib_dir_VALUE'
 auth_validation_public_key_path = 'auth_validation_public_key_path_VALUE'
 auth_type = 'auth_type_VALUE'
@@ -659,6 +675,7 @@ local_path = 'relish_storage_local_VALUE'
             checkpoint_period: Some("checkpoint_period_VALUE".to_string()),
             gc_horizon: Some("gc_horizon_VALUE".to_string()),
             gc_period: Some("gc_period_VALUE".to_string()),
+            open_mem_limit: Some("open_mem_limit_VALUE".to_string()),
             pg_distrib_dir: Some("pg_distrib_dir_VALUE".to_string()),
             auth_validation_public_key_path: Some(
                 "auth_validation_public_key_path_VALUE".to_string(),
@@ -685,6 +702,7 @@ checkpoint_distance = 'checkpoint_distance_VALUE'
 checkpoint_period = 'checkpoint_period_VALUE'
 gc_horizon = 'gc_horizon_VALUE'
 gc_period = 'gc_period_VALUE'
+open_mem_limit = 'open_mem_limit_VALUE'
 pg_distrib_dir = 'pg_distrib_dir_VALUE'
 auth_validation_public_key_path = 'auth_validation_public_key_path_VALUE'
 auth_type = 'auth_type_VALUE'

pageserver/src/branches.rs

@@ -23,6 +23,7 @@ use zenith_utils::zid::{ZTenantId, ZTimelineId};
 
 use crate::tenant_mgr;
 use crate::walredo::WalRedoManager;
+use crate::CheckpointConfig;
 use crate::{repository::Repository, PageServerConf};
 use crate::{restore_local_repo, LOG_FILE_NAME};
 
@@ -35,7 +36,7 @@ pub struct BranchInfo {
     pub ancestor_id: Option<String>,
     pub ancestor_lsn: Option<String>,
     pub current_logical_size: usize,
-    pub current_logical_size_non_incremental: usize,
+    pub current_logical_size_non_incremental: Option<usize>,
 }
 
 impl BranchInfo {
@@ -44,6 +45,7 @@ impl BranchInfo {
         conf: &PageServerConf,
         tenantid: &ZTenantId,
         repo: &Arc<dyn Repository>,
+        include_non_incremental_logical_size: bool,
     ) -> Result<Self> {
         let name = path
             .as_ref()
@@ -78,6 +80,14 @@ impl BranchInfo {
             );
         }
 
+        // non incremental size calculation can be heavy, so let it be optional
+        // needed for tests to check size calculation
+        let current_logical_size_non_incremental = include_non_incremental_logical_size
+            .then(|| {
+                timeline.get_current_logical_size_non_incremental(timeline.get_last_record_lsn())
+            })
+            .transpose()?;
+
         Ok(BranchInfo {
             name,
             timeline_id,
@@ -85,8 +95,7 @@ impl BranchInfo {
             ancestor_id,
             ancestor_lsn,
             current_logical_size: timeline.get_current_logical_size(),
-            current_logical_size_non_incremental: timeline
-                .get_current_logical_size_non_incremental(timeline.get_last_record_lsn())?,
+            current_logical_size_non_incremental,
         })
     }
 }
@@ -230,7 +239,7 @@ fn bootstrap_timeline(
         timeline.writer().as_ref(),
         lsn,
     )?;
-    timeline.checkpoint()?;
+    timeline.checkpoint(CheckpointConfig::Forced)?;
 
     println!(
         "created initial timeline {} timeline.lsn {}",
@@ -248,7 +257,11 @@ fn bootstrap_timeline(
     Ok(())
 }
 
-pub(crate) fn get_branches(conf: &PageServerConf, tenantid: &ZTenantId) -> Result<Vec<BranchInfo>> {
+pub(crate) fn get_branches(
+    conf: &PageServerConf,
+    tenantid: &ZTenantId,
+    include_non_incremental_logical_size: bool,
+) -> Result<Vec<BranchInfo>> {
     let repo = tenant_mgr::get_repository_for_tenant(*tenantid)?;
 
     // Each branch has a corresponding record (text file) in the refs/branches
@@ -258,7 +271,13 @@ pub(crate) fn get_branches(conf: &PageServerConf, tenantid: &ZTenantId) -> Resul
     std::fs::read_dir(&branches_dir)?
         .map(|dir_entry_res| {
             let dir_entry = dir_entry_res?;
-            BranchInfo::from_path(dir_entry.path(), conf, tenantid, &repo)
+            BranchInfo::from_path(
+                dir_entry.path(),
+                conf,
+                tenantid,
+                &repo,
+                include_non_incremental_logical_size,
+            )
         })
         .collect()
 }
@@ -320,7 +339,7 @@ pub(crate) fn create_branch(
         ancestor_id: None,
         ancestor_lsn: None,
         current_logical_size: 0,
-        current_logical_size_non_incremental: 0,
+        current_logical_size_non_incremental: Some(0),
     })
 }
 

pageserver/src/http/openapi_spec.yml

@@ -25,6 +25,11 @@ paths:
         schema:
           type: string
           format: hex
+      - name: include-non-incremental-logical-size
+        in: query
+        schema:
+          type: string
+          description: Controls calculation of current_logical_size_non_incremental
     get:
       description: Get branches for tenant
       responses:
@@ -73,6 +78,11 @@ paths:
         required: true
         schema:
           type: string
+      - name: include-non-incremental-logical-size
+        in: query
+        schema:
+          type: string
+          description: Controls calculation of current_logical_size_non_incremental
     get:
       description: Get branches for tenant
       responses:
@@ -260,7 +270,6 @@ components:
         - timeline_id
         - latest_valid_lsn
         - current_logical_size
-        - current_logical_size_non_incremental
       properties:
         name:
           type: string

pageserver/src/http/routes.rs

@@ -86,31 +86,59 @@ async fn branch_create_handler(mut request: Request<Body>) -> Result<Response<Bo
     Ok(json_response(StatusCode::CREATED, response_data)?)
 }
 
+// Gate non incremental logical size calculation behind a flag
+// after pgbench -i -s100 calculation took 28ms so if multiplied by the number of timelines
+// and tenants it can take noticeable amount of time. Also the value currently used only in tests
+fn get_include_non_incremental_logical_size(request: &Request<Body>) -> bool {
+    request
+        .uri()
+        .query()
+        .map(|v| {
+            url::form_urlencoded::parse(v.as_bytes())
+                .into_owned()
+                .any(|(param, _)| param == "include-non-incremental-logical-size")
+        })
+        .unwrap_or(false)
+}
+
 async fn branch_list_handler(request: Request<Body>) -> Result<Response<Body>, ApiError> {
     let tenantid: ZTenantId = parse_request_param(&request, "tenant_id")?;
 
+    let include_non_incremental_logical_size = get_include_non_incremental_logical_size(&request);
+
     check_permission(&request, Some(tenantid))?;
 
     let response_data = tokio::task::spawn_blocking(move || {
         let _enter = info_span!("branch_list", tenant = %tenantid).entered();
-        crate::branches::get_branches(get_config(&request), &tenantid)
+        crate::branches::get_branches(
+            get_config(&request),
+            &tenantid,
+            include_non_incremental_logical_size,
+        )
     })
     .await
     .map_err(ApiError::from_err)??;
     Ok(json_response(StatusCode::OK, response_data)?)
 }
 
-// TODO add to swagger
 async fn branch_detail_handler(request: Request<Body>) -> Result<Response<Body>, ApiError> {
     let tenantid: ZTenantId = parse_request_param(&request, "tenant_id")?;
     let branch_name: String = get_request_param(&request, "branch_name")?.to_string();
     let conf = get_state(&request).conf;
     let path = conf.branch_path(&branch_name, &tenantid);
 
+    let include_non_incremental_logical_size = get_include_non_incremental_logical_size(&request);
+
     let response_data = tokio::task::spawn_blocking(move || {
         let _enter = info_span!("branch_detail", tenant = %tenantid, branch=%branch_name).entered();
         let repo = tenant_mgr::get_repository_for_tenant(tenantid)?;
-        BranchInfo::from_path(path, conf, &tenantid, &repo)
+        BranchInfo::from_path(
+            path,
+            conf,
+            &tenantid,
+            &repo,
+            include_non_incremental_logical_size,
+        )
     })
     .await
     .map_err(ApiError::from_err)??;

pageserver/src/layered_repository.rs

@@ -16,13 +16,12 @@ use bookfile::Book;
 use bytes::Bytes;
 use lazy_static::lazy_static;
 use postgres_ffi::pg_constants::BLCKSZ;
-use serde::{Deserialize, Serialize};
 use tracing::*;
+use zenith_utils::batch_fsync::BatchFsync;
 
 use std::collections::hash_map::Entry;
 use std::collections::HashMap;
 use std::collections::{BTreeSet, HashSet};
-use std::convert::TryInto;
 use std::fs;
 use std::fs::{File, OpenOptions};
 use std::io::Write;
@@ -30,9 +29,9 @@ use std::ops::{Bound::Included, Deref};
 use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex, MutexGuard};
-use std::thread::JoinHandle;
 use std::time::{Duration, Instant};
 
+use self::metadata::{metadata_path, TimelineMetadata};
 use crate::relish::*;
 use crate::relish_storage::schedule_timeline_upload;
 use crate::repository::{GcResult, Repository, Timeline, TimelineWriter, WALRecord};
@@ -40,6 +39,7 @@ use crate::tenant_mgr;
 use crate::walreceiver;
 use crate::walreceiver::IS_WAL_RECEIVER;
 use crate::walredo::WalRedoManager;
+use crate::CheckpointConfig;
 use crate::PageServerConf;
 use crate::{ZTenantId, ZTimelineId};
 
@@ -47,7 +47,6 @@ use zenith_metrics::{
     register_histogram, register_int_gauge_vec, Histogram, IntGauge, IntGaugeVec,
 };
 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;
@@ -55,16 +54,19 @@ use zenith_utils::seqwait::SeqWait;
 mod blob;
 mod delta_layer;
 mod filename;
+mod global_layer_map;
 mod image_layer;
 mod inmemory_layer;
 mod interval_tree;
 mod layer_map;
+pub mod metadata;
 mod page_versions;
 mod storage_layer;
 
 use delta_layer::DeltaLayer;
 use image_layer::ImageLayer;
 
+use global_layer_map::{LayerId, GLOBAL_LAYER_MAP};
 use inmemory_layer::InMemoryLayer;
 use layer_map::LayerMap;
 use storage_layer::{
@@ -111,8 +113,6 @@ lazy_static! {
     .expect("failed to define a metric");
 }
 
-/// The name of the metadata file pageserver creates per timeline.
-pub const METADATA_FILE_NAME: &str = "metadata";
 /// Parts of the `.zenith/tenants/<tenantid>/timelines/<timelineid>` directory prefix.
 pub const TENANTS_SEGMENT_NAME: &str = "tenants";
 pub const TIMELINES_SEGMENT_NAME: &str = "timelines";
@@ -145,12 +145,7 @@ impl Repository for LayeredRepository {
         // Create the timeline directory, and write initial metadata to file.
         crashsafe_dir::create_dir_all(self.conf.timeline_path(&timelineid, &self.tenantid))?;
 
-        let metadata = TimelineMetadata {
-            disk_consistent_lsn: Lsn(0),
-            prev_record_lsn: None,
-            ancestor_timeline: None,
-            ancestor_lsn: Lsn(0),
-        };
+        let metadata = TimelineMetadata::new(Lsn(0), None, None, Lsn(0));
         Self::save_metadata(self.conf, timelineid, self.tenantid, &metadata, true)?;
 
         let timeline = LayeredTimeline::new(
@@ -189,12 +184,7 @@ impl Repository for LayeredRepository {
         // Create the metadata file, noting the ancestor of the new timeline.
         // There is initially no data in it, but all the read-calls know to look
         // into the ancestor.
-        let metadata = TimelineMetadata {
-            disk_consistent_lsn: start_lsn,
-            prev_record_lsn: dst_prev,
-            ancestor_timeline: Some(src),
-            ancestor_lsn: start_lsn,
-        };
+        let metadata = TimelineMetadata::new(start_lsn, dst_prev, Some(src), start_lsn);
         crashsafe_dir::create_dir_all(self.conf.timeline_path(&dst, &self.tenantid))?;
         Self::save_metadata(self.conf, dst, self.tenantid, &metadata, true)?;
 
@@ -219,6 +209,22 @@ impl Repository for LayeredRepository {
             })
     }
 
+    fn checkpoint_iteration(&self, cconf: CheckpointConfig) -> Result<()> {
+        {
+            let timelines = self.timelines.lock().unwrap();
+
+            for (timelineid, timeline) in timelines.iter() {
+                let _entered =
+                    info_span!("checkpoint", timeline = %timelineid, tenant = %self.tenantid)
+                        .entered();
+
+                timeline.checkpoint(cconf)?;
+            }
+        }
+
+        Ok(())
+    }
+
     // Wait for all threads to complete and persist repository data before pageserver shutdown.
     fn shutdown(&self) -> Result<()> {
         trace!("LayeredRepository shutdown for tenant {}", self.tenantid);
@@ -228,7 +234,7 @@ impl Repository for LayeredRepository {
             walreceiver::stop_wal_receiver(*timelineid);
             // Wait for syncing data to disk
             trace!("repo shutdown. checkpoint timeline {}", timelineid);
-            timeline.checkpoint()?;
+            timeline.checkpoint(CheckpointConfig::Forced)?;
 
             //TODO Wait for walredo process to shutdown too
         }
@@ -250,14 +256,14 @@ 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;
+                let disk_consistent_lsn = metadata.disk_consistent_lsn();
 
                 // Recurse to look up the ancestor timeline.
                 //
                 // TODO: If you have a very deep timeline history, this could become
                 // expensive. Perhaps delay this until we need to look up a page in
                 // ancestor.
-                let ancestor = if let Some(ancestor_timelineid) = metadata.ancestor_timeline {
+                let ancestor = if let Some(ancestor_timelineid) = metadata.ancestor_timeline() {
                     Some(self.get_timeline_locked(ancestor_timelineid, timelines)?)
                 } else {
                     None
@@ -309,90 +315,6 @@ impl LayeredRepository {
         }
     }
 
-    ///
-    /// Launch the checkpointer thread in given repository.
-    ///
-    pub fn launch_checkpointer_thread(
-        conf: &'static PageServerConf,
-        rc: Arc<LayeredRepository>,
-    ) -> JoinHandle<()> {
-        std::thread::Builder::new()
-            .name("Checkpointer thread".into())
-            .spawn(move || {
-                // FIXME: relaunch it? Panic is not good.
-                rc.checkpoint_loop(conf).expect("Checkpointer thread died");
-            })
-            .unwrap()
-    }
-
-    ///
-    /// Checkpointer thread's main loop
-    ///
-    fn checkpoint_loop(&self, conf: &'static PageServerConf) -> Result<()> {
-        while !tenant_mgr::shutdown_requested() {
-            std::thread::sleep(conf.checkpoint_period);
-            info!("checkpointer thread for tenant {} waking up", self.tenantid);
-
-            // checkpoint timelines that have accumulated more than CHECKPOINT_DISTANCE
-            // bytes of WAL since last checkpoint.
-            {
-                let timelines = self.timelines.lock().unwrap();
-                for (timelineid, timeline) in timelines.iter() {
-                    let _entered =
-                        info_span!("checkpoint", timeline = %timelineid, tenant = %self.tenantid)
-                            .entered();
-
-                    STORAGE_TIME
-                        .with_label_values(&["checkpoint_timed"])
-                        .observe_closure_duration(|| {
-                            timeline.checkpoint_internal(conf.checkpoint_distance, false)
-                        })?
-                }
-                // release lock on 'timelines'
-            }
-        }
-        trace!("Checkpointer thread shut down");
-        Ok(())
-    }
-
-    ///
-    /// Launch the GC thread in given repository.
-    ///
-    pub fn launch_gc_thread(
-        conf: &'static PageServerConf,
-        rc: Arc<LayeredRepository>,
-    ) -> JoinHandle<()> {
-        std::thread::Builder::new()
-            .name("GC thread".into())
-            .spawn(move || {
-                // FIXME: relaunch it? Panic is not good.
-                rc.gc_loop(conf).expect("GC thread died");
-            })
-            .unwrap()
-    }
-
-    ///
-    /// GC thread's main loop
-    ///
-    fn gc_loop(&self, conf: &'static PageServerConf) -> Result<()> {
-        while !tenant_mgr::shutdown_requested() {
-            // Garbage collect old files that are not needed for PITR anymore
-            if conf.gc_horizon > 0 {
-                self.gc_iteration(None, conf.gc_horizon, false).unwrap();
-            }
-
-            // TODO Write it in more adequate way using
-            // condvar.wait_timeout() or something
-            let mut sleep_time = conf.gc_period.as_secs();
-            while sleep_time > 0 && !tenant_mgr::shutdown_requested() {
-                sleep_time -= 1;
-                std::thread::sleep(Duration::from_secs(1));
-            }
-            info!("gc thread for tenant {} waking up", self.tenantid);
-        }
-        Ok(())
-    }
-
     /// Save timeline metadata to file
     fn save_metadata(
         conf: &'static PageServerConf,
@@ -546,7 +468,7 @@ impl LayeredRepository {
                 // so that they too can be garbage collected. That's
                 // used in tests, so we want as deterministic results as possible.
                 if checkpoint_before_gc {
-                    timeline.checkpoint()?;
+                    timeline.checkpoint(CheckpointConfig::Forced)?;
                     info!("timeline {} checkpoint_before_gc done", timelineid);
                 }
 
@@ -561,66 +483,6 @@ impl LayeredRepository {
     }
 }
 
-/// Metadata stored on disk for each timeline
-///
-/// The fields correspond to the values we hold in memory, in LayeredTimeline.
-#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
-pub struct TimelineMetadata {
-    /// [`Lsn`] that corresponds to the corresponding timeline directory
-    /// contents, stored locally in the pageserver workdir.
-    pub disk_consistent_lsn: Lsn,
-
-    // This is only set if we know it. We track it in memory when the page
-    // server is running, but we only track the value corresponding to
-    // 'last_record_lsn', not 'disk_consistent_lsn' which can lag behind by a
-    // lot. We only store it in the metadata file when we flush *all* the
-    // in-memory data so that 'last_record_lsn' is the same as
-    // 'disk_consistent_lsn'.  That's OK, because after page server restart, as
-    // soon as we reprocess at least one record, we will have a valid
-    // 'prev_record_lsn' value in memory again. This is only really needed when
-    // doing a clean shutdown, so that there is no more WAL beyond
-    // 'disk_consistent_lsn'
-    pub prev_record_lsn: Option<Lsn>,
-
-    pub ancestor_timeline: Option<ZTimelineId>,
-    pub ancestor_lsn: Lsn,
-}
-
-impl TimelineMetadata {
-    pub fn from_bytes(metadata_bytes: &[u8]) -> anyhow::Result<Self> {
-        ensure!(
-            metadata_bytes.len() == METADATA_MAX_SAFE_SIZE,
-            "metadata bytes size is wrong"
-        );
-
-        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,
-            "metadata checksum mismatch"
-        );
-
-        let data = TimelineMetadata::des_prefix(data)?;
-        assert!(data.disk_consistent_lsn.is_aligned());
-
-        Ok(data)
-    }
-
-    pub fn to_bytes(&self) -> anyhow::Result<Vec<u8>> {
-        let mut metadata_bytes = TimelineMetadata::ser(self)?;
-        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));
-        Ok(metadata_bytes)
-    }
-}
-
 pub struct LayeredTimeline {
     conf: &'static PageServerConf,
 
@@ -867,11 +729,15 @@ impl Timeline for LayeredTimeline {
     /// Public entry point for checkpoint(). All the logic is in the private
     /// checkpoint_internal function, this public facade just wraps it for
     /// metrics collection.
-    fn checkpoint(&self) -> Result<()> {
-        STORAGE_TIME
-            .with_label_values(&["checkpoint_force"])
-            //pass checkpoint_distance=0 to force checkpoint
-            .observe_closure_duration(|| self.checkpoint_internal(0, true))
+    fn checkpoint(&self, cconf: CheckpointConfig) -> Result<()> {
+        match cconf {
+            CheckpointConfig::Forced => STORAGE_TIME
+                .with_label_values(&["forced checkpoint"])
+                .observe_closure_duration(|| self.checkpoint_internal(0)),
+            CheckpointConfig::Distance(distance) => STORAGE_TIME
+                .with_label_values(&["checkpoint"])
+                .observe_closure_duration(|| self.checkpoint_internal(distance)),
+        }
     }
 
     fn get_last_record_lsn(&self) -> Lsn {
@@ -935,6 +801,10 @@ impl Timeline for LayeredTimeline {
             _write_guard: self.write_lock.lock().unwrap(),
         })
     }
+
+    fn upgrade_to_layered_timeline(&self) -> &crate::layered_repository::LayeredTimeline {
+        self
+    }
 }
 
 impl LayeredTimeline {
@@ -965,13 +835,13 @@ impl LayeredTimeline {
 
             // initialize in-memory 'last_record_lsn' from 'disk_consistent_lsn'.
             last_record_lsn: SeqWait::new(RecordLsn {
-                last: metadata.disk_consistent_lsn,
-                prev: metadata.prev_record_lsn.unwrap_or(Lsn(0)),
+                last: metadata.disk_consistent_lsn(),
+                prev: metadata.prev_record_lsn().unwrap_or(Lsn(0)),
             }),
-            disk_consistent_lsn: AtomicLsn::new(metadata.disk_consistent_lsn.0),
+            disk_consistent_lsn: AtomicLsn::new(metadata.disk_consistent_lsn().0),
 
             ancestor_timeline: ancestor,
-            ancestor_lsn: metadata.ancestor_lsn,
+            ancestor_lsn: metadata.ancestor_lsn(),
             current_logical_size: AtomicUsize::new(current_logical_size),
             current_logical_size_gauge,
             upload_relishes,
@@ -1250,7 +1120,7 @@ impl LayeredTimeline {
     /// Flush to disk all data that was written with the put_* functions
     ///
     /// NOTE: This has nothing to do with checkpoint in PostgreSQL.
-    fn checkpoint_internal(&self, checkpoint_distance: u64, forced: bool) -> Result<()> {
+    fn checkpoint_internal(&self, checkpoint_distance: u64) -> Result<()> {
         let mut write_guard = self.write_lock.lock().unwrap();
         let mut layers = self.layers.lock().unwrap();
 
@@ -1276,14 +1146,13 @@ 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;
-        let mut layer_uploads = Vec::new();
-        while let Some((oldest_layer, oldest_generation)) = layers.peek_oldest_open() {
-            let oldest_pending_lsn = oldest_layer.get_oldest_pending_lsn();
+        let mut batch_fsync = BatchFsync::default();
 
-            if tenant_mgr::shutdown_requested() && !forced {
-                return Ok(());
-            }
+        let mut layer_uploads = Vec::new();
+        while let Some((oldest_layer_id, oldest_layer, oldest_generation)) =
+            layers.peek_oldest_open()
+        {
+            let oldest_pending_lsn = oldest_layer.get_oldest_pending_lsn();
 
             // Does this layer need freezing?
             //
@@ -1307,42 +1176,14 @@ impl LayeredTimeline {
                 break;
             }
 
-            // Mark the layer as no longer accepting writes and record the end_lsn.
-            // This happens in-place, no new layers are created now.
-            // We call `get_last_record_lsn` again, which may be different from the
-            // original load, as we may have released the write lock since then.
-            oldest_layer.freeze(self.get_last_record_lsn());
-
-            // The layer is no longer open, update the layer map to reflect this.
-            // We will replace it with on-disk historics below.
-            layers.pop_oldest_open();
-            layers.insert_historic(oldest_layer.clone());
-
-            // Write the now-frozen layer to disk. That could take a while, so release the lock while do it
             drop(layers);
             drop(write_guard);
 
-            let new_historics = oldest_layer.write_to_disk(self)?;
+            let mut this_layer_uploads = self.evict_layer(oldest_layer_id, &mut batch_fsync)?;
+            layer_uploads.append(&mut this_layer_uploads);
 
             write_guard = self.write_lock.lock().unwrap();
             layers = self.layers.lock().unwrap();
-
-            if !new_historics.is_empty() {
-                created_historics = true;
-            }
-
-            // Finally, replace the frozen in-memory layer with the new on-disk layers
-            layers.remove_historic(oldest_layer);
-
-            // Add the historics to the LayerMap
-            for delta_layer in new_historics.delta_layers {
-                layer_uploads.push(delta_layer.path());
-                layers.insert_historic(Arc::new(delta_layer));
-            }
-            for image_layer in new_historics.image_layers {
-                layer_uploads.push(image_layer.path());
-                layers.insert_historic(Arc::new(image_layer));
-            }
         }
 
         // Call unload() on all frozen layers, to release memory.
@@ -1355,14 +1196,16 @@ impl LayeredTimeline {
         drop(layers);
         drop(write_guard);
 
-        if created_historics {
+        if !layer_uploads.is_empty() {
             // We must fsync the timeline dir to ensure the directory entries for
             // new layer files are durable
             let timeline_dir =
                 File::open(self.conf.timeline_path(&self.timelineid, &self.tenantid))?;
-            timeline_dir.sync_all()?;
+            batch_fsync.add(timeline_dir)?;
         }
 
+        batch_fsync.done()?;
+
         // If we were able to advance 'disk_consistent_lsn', save it the metadata file.
         // After crash, we will restart WAL streaming and processing from that point.
         let old_disk_consistent_lsn = self.disk_consistent_lsn.load();
@@ -1383,12 +1226,12 @@ impl LayeredTimeline {
 
             let ancestor_timelineid = self.ancestor_timeline.as_ref().map(|x| x.timelineid);
 
-            let metadata = TimelineMetadata {
+            let metadata = TimelineMetadata::new(
                 disk_consistent_lsn,
-                prev_record_lsn: ondisk_prev_record_lsn,
-                ancestor_timeline: ancestor_timelineid,
-                ancestor_lsn: self.ancestor_lsn,
-            };
+                ondisk_prev_record_lsn,
+                ancestor_timelineid,
+                self.ancestor_lsn,
+            );
 
             LayeredRepository::save_metadata(
                 self.conf,
@@ -1408,6 +1251,55 @@ impl LayeredTimeline {
         Ok(())
     }
 
+    fn evict_layer(&self, layer_id: LayerId, batch_fsync: &mut BatchFsync) -> Result<Vec<PathBuf>> {
+        // Mark the layer as no longer accepting writes and record the end_lsn.
+        // This happens in-place, no new layers are created now.
+        // We call `get_last_record_lsn` again, which may be different from the
+        // original load, as we may have released the write lock since then.
+
+        let mut write_guard = self.write_lock.lock().unwrap();
+        let mut layers = self.layers.lock().unwrap();
+
+        let mut layer_uploads = Vec::new();
+
+        let global_layer_map = GLOBAL_LAYER_MAP.read().unwrap();
+        if let Some(oldest_layer) = global_layer_map.get(&layer_id) {
+            drop(global_layer_map);
+            oldest_layer.freeze(self.get_last_record_lsn());
+
+            // The layer is no longer open, update the layer map to reflect this.
+            // We will replace it with on-disk historics below.
+            layers.remove_open(layer_id);
+            layers.insert_historic(oldest_layer.clone());
+
+            // Write the now-frozen layer to disk. That could take a while, so release the lock while do it
+            drop(layers);
+            drop(write_guard);
+
+            let new_historics = oldest_layer.write_to_disk(self, batch_fsync)?;
+
+            write_guard = self.write_lock.lock().unwrap();
+            layers = self.layers.lock().unwrap();
+
+            // Finally, replace the frozen in-memory layer with the new on-disk layers
+            layers.remove_historic(oldest_layer);
+
+            // Add the historics to the LayerMap
+            for delta_layer in new_historics.delta_layers {
+                layer_uploads.push(delta_layer.path());
+                layers.insert_historic(Arc::new(delta_layer));
+            }
+            for image_layer in new_historics.image_layers {
+                layer_uploads.push(image_layer.path());
+                layers.insert_historic(Arc::new(image_layer));
+            }
+        }
+        drop(layers);
+        drop(write_guard);
+
+        Ok(layer_uploads)
+    }
+
     ///
     /// Garbage collect layer files on a timeline that are no longer needed.
     ///
@@ -1956,15 +1848,6 @@ pub fn dump_layerfile_from_path(path: &Path) -> Result<()> {
     Ok(())
 }
 
-pub fn metadata_path(
-    conf: &'static PageServerConf,
-    timelineid: ZTimelineId,
-    tenantid: ZTenantId,
-) -> PathBuf {
-    conf.timeline_path(&timelineid, &tenantid)
-        .join(METADATA_FILE_NAME)
-}
-
 /// 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<()> {
@@ -1984,3 +1867,36 @@ fn rename_to_backup(path: PathBuf) -> anyhow::Result<()> {
         path
     ))
 }
+
+//----- Global layer management
+
+/// Check if too much memory is being used by open layers. If so, evict
+pub fn evict_layer_if_needed(conf: &PageServerConf) -> Result<()> {
+    // Keep evicting layers until we are below the memory threshold.
+    let mut global_layer_map = GLOBAL_LAYER_MAP.read().unwrap();
+    while let Some((layer_id, layer)) = global_layer_map.find_victim_if_needed(conf.open_mem_limit)
+    {
+        drop(global_layer_map);
+        let tenantid = layer.get_tenant_id();
+        let timelineid = layer.get_timeline_id();
+
+        let _entered =
+            info_span!("global evict", timeline = %timelineid, tenant = %tenantid).entered();
+        info!("evicting {}", layer.filename().display());
+        drop(layer);
+
+        let timeline = tenant_mgr::get_timeline_for_tenant(tenantid, timelineid)?;
+
+        let mut batch_fsync = BatchFsync::default();
+
+        timeline
+            .upgrade_to_layered_timeline()
+            .evict_layer(layer_id, &mut batch_fsync)?;
+
+        batch_fsync.done()?;
+
+        global_layer_map = GLOBAL_LAYER_MAP.read().unwrap();
+    }
+
+    Ok(())
+}

pageserver/src/layered_repository/delta_layer.rs

@@ -48,6 +48,7 @@ use crate::{ZTenantId, ZTimelineId};
 use anyhow::{bail, ensure, Result};
 use log::*;
 use serde::{Deserialize, Serialize};
+use zenith_utils::batch_fsync::BatchFsync;
 use zenith_utils::vec_map::VecMap;
 // avoid binding to Write (conflicts with std::io::Write)
 // while being able to use std::fmt::Write's methods
@@ -148,6 +149,10 @@ pub struct DeltaLayerInner {
 }
 
 impl Layer for DeltaLayer {
+    fn get_tenant_id(&self) -> ZTenantId {
+        self.tenantid
+    }
+
     fn get_timeline_id(&self) -> ZTimelineId {
         self.timelineid
     }
@@ -360,6 +365,7 @@ impl DeltaLayer {
         dropped: bool,
         page_versions: impl Iterator<Item = (u32, Lsn, &'a PageVersion)>,
         relsizes: VecMap<Lsn, u32>,
+        batch_fsync: &mut BatchFsync,
     ) -> Result<DeltaLayer> {
         if seg.rel.is_blocky() {
             assert!(!relsizes.is_empty());
@@ -432,7 +438,8 @@ impl DeltaLayer {
 
         // This flushes the underlying 'buf_writer'.
         let writer = book.close()?;
-        writer.get_ref().sync_all()?;
+        let file = writer.into_inner()?;
+        batch_fsync.add(file)?;
 
         trace!("saved {}", &path.display());
 

pageserver/src/layered_repository/filename.rs

@@ -13,7 +13,7 @@ use anyhow::Result;
 use log::*;
 use zenith_utils::lsn::Lsn;
 
-use super::METADATA_FILE_NAME;
+use super::metadata::METADATA_FILE_NAME;
 
 // Note: LayeredTimeline::load_layer_map() relies on this sort order
 #[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Clone)]

pageserver/src/layered_repository/global_layer_map.rs

@@ -0,0 +1,209 @@
+//!
+//! Global registry of open layers.
+//!
+//! Whenever a new in-memory layer is created to hold incoming WAL, it is registered
+//! in [`GLOBAL_LAYER_MAP`], so that we can keep track of the total number of
+//! in-memory layers in the system, and know when we need to evict some to release
+//! memory.
+//!
+//! Each layer is assigned a unique ID when it's registered in the global registry.
+//! The ID can be used to relocate the layer later, without having to hold locks.
+//!
+
+use std::sync::atomic::{AtomicU8, AtomicUsize, Ordering};
+use std::sync::{Arc, RwLock};
+
+use super::inmemory_layer::InMemoryLayer;
+
+use lazy_static::lazy_static;
+
+const MAX_USAGE_COUNT: u8 = 5;
+
+lazy_static! {
+    pub static ref GLOBAL_LAYER_MAP: RwLock<OpenLayers> = RwLock::new(OpenLayers::default());
+}
+
+///
+/// How much memory is being used by all the open layers? This is used to trigger
+/// freezing and evicting an open layer to disk.
+///
+/// This is only a rough approximation, it leaves out a lot of things like malloc()
+/// overhead. But as long there is enough "slop" and it's not set too close to the RAM
+/// size on the system, it's good enough.
+pub static GLOBAL_OPEN_MEM_USAGE: AtomicUsize = AtomicUsize::new(0);
+
+// TODO these types can probably be smaller
+#[derive(PartialEq, Eq, Clone, Copy)]
+pub struct LayerId {
+    index: usize,
+    tag: u64, // to avoid ABA problem
+}
+
+enum SlotData {
+    Occupied(Arc<InMemoryLayer>),
+    /// Vacant slots form a linked list, the value is the index
+    /// of the next vacant slot in the list.
+    Vacant(Option<usize>),
+}
+
+struct Slot {
+    tag: u64,
+    data: SlotData,
+    usage_count: AtomicU8, // for clock algorithm
+}
+
+#[derive(Default)]
+pub struct OpenLayers {
+    slots: Vec<Slot>,
+    num_occupied: usize,
+    next_victim: AtomicUsize,
+
+    // Head of free-slot list.
+    next_empty_slot_idx: Option<usize>,
+}
+
+impl OpenLayers {
+    pub fn insert(&mut self, layer: Arc<InMemoryLayer>) -> LayerId {
+        let slot_idx = match self.next_empty_slot_idx {
+            Some(slot_idx) => slot_idx,
+            None => {
+                let idx = self.slots.len();
+                self.slots.push(Slot {
+                    tag: 0,
+                    data: SlotData::Vacant(None),
+                    usage_count: AtomicU8::new(0),
+                });
+                idx
+            }
+        };
+        let slots_len = self.slots.len();
+
+        let slot = &mut self.slots[slot_idx];
+
+        match slot.data {
+            SlotData::Occupied(_) => {
+                panic!("an occupied slot was in the free list");
+            }
+            SlotData::Vacant(next_empty_slot_idx) => {
+                self.next_empty_slot_idx = next_empty_slot_idx;
+            }
+        }
+
+        slot.data = SlotData::Occupied(layer);
+        slot.usage_count.store(1, Ordering::Relaxed);
+
+        self.num_occupied += 1;
+        assert!(self.num_occupied <= slots_len);
+
+        LayerId {
+            index: slot_idx,
+            tag: slot.tag,
+        }
+    }
+
+    pub fn get(&self, layer_id: &LayerId) -> Option<Arc<InMemoryLayer>> {
+        let slot = self.slots.get(layer_id.index)?; // TODO should out of bounds indexes just panic?
+        if slot.tag != layer_id.tag {
+            return None;
+        }
+
+        if let SlotData::Occupied(layer) = &slot.data {
+            let _ = slot.usage_count.fetch_update(
+                Ordering::Relaxed,
+                Ordering::Relaxed,
+                |old_usage_count| {
+                    if old_usage_count < MAX_USAGE_COUNT {
+                        Some(old_usage_count + 1)
+                    } else {
+                        None
+                    }
+                },
+            );
+            Some(Arc::clone(layer))
+        } else {
+            None
+        }
+    }
+
+    /// Find a victim layer to evict, if the total memory usage of all open layers
+    /// is larger than 'limit'
+    pub fn find_victim_if_needed(&self, limit: usize) -> Option<(LayerId, Arc<InMemoryLayer>)> {
+        let mem_usage = GLOBAL_OPEN_MEM_USAGE.load(Ordering::Relaxed);
+
+        if mem_usage > limit {
+            self.find_victim()
+        } else {
+            None
+        }
+    }
+
+    pub fn find_victim(&self) -> Option<(LayerId, Arc<InMemoryLayer>)> {
+        if self.num_occupied == 0 {
+            return None;
+        }
+
+        // Run the clock algorithm.
+        //
+        // FIXME: It's theoretically possible that a constant stream of get() requests
+        // comes in faster than we advance the clock hand, so that this never finishes.
+        loop {
+            // FIXME: Because we interpret the clock hand variable modulo slots.len(), the
+            // hand effectively jumps to a more or less random place whenever the array is
+            // expanded. That's relatively harmless, it just leads to a non-optimal choice
+            // of victim. Also, in a server that runs for long enough, the array should reach
+            // a steady-state size and not grow anymore.
+            let next_victim = self.next_victim.fetch_add(1, Ordering::Relaxed) % self.slots.len();
+
+            let slot = &self.slots[next_victim];
+
+            if let SlotData::Occupied(data) = &slot.data {
+                fn update_fn(old_usage_count: u8) -> Option<u8> {
+                    if old_usage_count > 0 {
+                        Some(old_usage_count - 1)
+                    } else {
+                        None
+                    }
+                }
+
+                if slot
+                    .usage_count
+                    .fetch_update(Ordering::Relaxed, Ordering::Relaxed, update_fn)
+                    .is_err()
+                {
+                    // Found a slot with usage_count == 0. Return it.
+                    return Some((
+                        LayerId {
+                            index: next_victim,
+                            tag: slot.tag,
+                        },
+                        Arc::clone(data),
+                    ));
+                }
+            }
+        }
+    }
+
+    // TODO this won't be a public API in the future
+    pub fn remove(&mut self, layer_id: &LayerId) {
+        let slot = &mut self.slots[layer_id.index];
+
+        if slot.tag != layer_id.tag {
+            return;
+        }
+
+        match &slot.data {
+            SlotData::Occupied(_layer) => {
+                // TODO evict the layer
+            }
+            SlotData::Vacant(_) => unimplemented!(),
+        }
+
+        slot.data = SlotData::Vacant(self.next_empty_slot_idx);
+        self.next_empty_slot_idx = Some(layer_id.index);
+
+        assert!(self.num_occupied > 0);
+        self.num_occupied -= 1;
+
+        slot.tag = slot.tag.wrapping_add(1);
+    }
+}

pageserver/src/layered_repository/image_layer.rs

@@ -39,6 +39,7 @@ use std::fs::File;
 use std::io::{BufWriter, Write};
 use std::path::{Path, PathBuf};
 use std::sync::{Mutex, MutexGuard};
+use zenith_utils::batch_fsync::BatchFsync;
 
 use bookfile::{Book, BookWriter};
 
@@ -117,6 +118,10 @@ impl Layer for ImageLayer {
         PathBuf::from(self.layer_name().to_string())
     }
 
+    fn get_tenant_id(&self) -> ZTenantId {
+        self.tenantid
+    }
+
     fn get_timeline_id(&self) -> ZTimelineId {
         self.timelineid
     }
@@ -257,6 +262,7 @@ impl ImageLayer {
         seg: SegmentTag,
         lsn: Lsn,
         base_images: Vec<Bytes>,
+        batch_fsync: &mut BatchFsync,
     ) -> Result<ImageLayer> {
         let image_type = if seg.rel.is_blocky() {
             let num_blocks: u32 = base_images.len().try_into()?;
@@ -316,7 +322,8 @@ impl ImageLayer {
 
         // This flushes the underlying 'buf_writer'.
         let writer = book.close()?;
-        writer.get_ref().sync_all()?;
+        let file = writer.into_inner()?;
+        batch_fsync.add(file)?;
 
         trace!("saved {}", path.display());
 
@@ -332,6 +339,7 @@ impl ImageLayer {
         timeline: &LayeredTimeline,
         src: &dyn Layer,
         lsn: Lsn,
+        batch_fsync: &mut BatchFsync,
     ) -> Result<ImageLayer> {
         let seg = src.get_seg_tag();
         let timelineid = timeline.timelineid;
@@ -360,7 +368,15 @@ impl ImageLayer {
             base_images.push(img);
         }
 
-        Self::create(conf, timelineid, timeline.tenantid, seg, lsn, base_images)
+        Self::create(
+            conf,
+            timelineid,
+            timeline.tenantid,
+            seg,
+            lsn,
+            base_images,
+            batch_fsync,
+        )
     }
 
     ///

pageserver/src/layered_repository/inmemory_layer.rs

@@ -3,6 +3,7 @@
 //! are held in a BTreeMap, and there's another BTreeMap to track the size of the relation.
 //!
 use crate::layered_repository::filename::DeltaFileName;
+use crate::layered_repository::global_layer_map::GLOBAL_OPEN_MEM_USAGE;
 use crate::layered_repository::storage_layer::{
     Layer, PageReconstructData, PageReconstructResult, PageVersion, SegmentTag, RELISH_SEG_SIZE,
 };
@@ -16,7 +17,9 @@ use anyhow::{bail, ensure, Result};
 use bytes::Bytes;
 use log::*;
 use std::path::PathBuf;
+use std::sync::atomic::Ordering;
 use std::sync::{Arc, RwLock};
+use zenith_utils::batch_fsync::BatchFsync;
 use zenith_utils::vec_map::VecMap;
 
 use zenith_utils::lsn::Lsn;
@@ -69,6 +72,15 @@ pub struct InMemoryLayerInner {
     /// a non-blocky rel, 'segsizes' is not used and is always empty.
     ///
     segsizes: VecMap<Lsn, u32>,
+
+    /// Approximate amount of memory used by this layer.
+    ///
+    /// TODO: This is currently a very crude metric, we don't take into account allocator
+    /// overhead, memory fragmentation, memory used by the VecMaps, nor many other things.
+    /// Just the actual # of bytes of a page image (8 kB) or the size of a WAL record.
+    ///
+    /// Whenever this is changed, you must also modify GLOBAL_OPEN_MEM_USAGE accordingly!
+    mem_usage: usize,
 }
 
 impl InMemoryLayerInner {
@@ -89,6 +101,15 @@ impl InMemoryLayerInner {
     }
 }
 
+impl Drop for InMemoryLayerInner {
+    fn drop(&mut self) {
+        if self.mem_usage > 0 {
+            GLOBAL_OPEN_MEM_USAGE.fetch_sub(self.mem_usage, Ordering::Relaxed);
+            self.mem_usage = 0;
+        }
+    }
+}
+
 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
@@ -113,6 +134,10 @@ impl Layer for InMemoryLayer {
         PathBuf::from(format!("inmem-{}", delta_filename))
     }
 
+    fn get_tenant_id(&self) -> ZTenantId {
+        self.tenantid
+    }
+
     fn get_timeline_id(&self) -> ZTimelineId {
         self.timelineid
     }
@@ -281,12 +306,6 @@ pub struct LayersOnDisk {
     pub image_layers: Vec<ImageLayer>,
 }
 
-impl LayersOnDisk {
-    pub fn is_empty(&self) -> bool {
-        self.delta_layers.is_empty() && self.image_layers.is_empty()
-    }
-}
-
 impl InMemoryLayer {
     /// Return the oldest page version that's stored in this layer
     pub fn get_oldest_pending_lsn(&self) -> Lsn {
@@ -330,6 +349,7 @@ impl InMemoryLayer {
                 dropped: false,
                 page_versions: PageVersions::default(),
                 segsizes,
+                mem_usage: 0,
             }),
         })
     }
@@ -376,6 +396,13 @@ impl InMemoryLayer {
 
         inner.assert_writeable();
 
+        let mut mem_usage = 0;
+        if let Some(img) = &pv.page_image {
+            mem_usage += img.len();
+        } else if let Some(rec) = &pv.record {
+            mem_usage += rec.rec.len();
+        }
+
         let old = inner.page_versions.append_or_update_last(blknum, lsn, pv);
 
         if old.is_some() {
@@ -386,6 +413,9 @@ impl InMemoryLayer {
             );
         }
 
+        inner.mem_usage += mem_usage;
+        GLOBAL_OPEN_MEM_USAGE.fetch_add(mem_usage, Ordering::Relaxed);
+
         // Also update the relation size, if this extended the relation.
         if self.seg.rel.is_blocky() {
             let newsize = blknum - self.seg.segno * RELISH_SEG_SIZE + 1;
@@ -521,6 +551,7 @@ impl InMemoryLayer {
                 dropped: false,
                 page_versions: PageVersions::default(),
                 segsizes,
+                mem_usage: 0,
             }),
         })
     }
@@ -550,6 +581,16 @@ impl InMemoryLayer {
             for (_blk, lsn, _pv) in inner.page_versions.ordered_page_version_iter(None) {
                 assert!(lsn <= end_lsn);
             }
+
+            // It's a bit premature to subtract the global mem usage here already.
+            // This layer consumes memory until it's written out to disk and dropped.
+            // But GLOBAL_OPEN_MEM_USAGE is used to trigger layer eviction, if there are
+            // too many open layers, and from that point of view this should no longer be
+            // counted against the global mem usage.
+            if inner.mem_usage > 0 {
+                GLOBAL_OPEN_MEM_USAGE.fetch_sub(inner.mem_usage, Ordering::Relaxed);
+                inner.mem_usage = 0;
+            }
         }
     }
 
@@ -561,7 +602,11 @@ impl InMemoryLayer {
     /// 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<LayersOnDisk> {
+    pub fn write_to_disk(
+        &self,
+        timeline: &LayeredTimeline,
+        batch_fsync: &mut BatchFsync,
+    ) -> Result<LayersOnDisk> {
         trace!(
             "write_to_disk {} get_end_lsn is {}",
             self.filename().display(),
@@ -591,6 +636,7 @@ impl InMemoryLayer {
                 true,
                 inner.page_versions.ordered_page_version_iter(None),
                 inner.segsizes.clone(),
+                batch_fsync,
             )?;
             trace!(
                 "freeze: created delta layer for dropped segment {} {}-{}",
@@ -628,6 +674,7 @@ impl InMemoryLayer {
                 false,
                 page_versions,
                 segsizes,
+                batch_fsync,
             )?;
             delta_layers.push(delta_layer);
             trace!(
@@ -644,7 +691,7 @@ impl InMemoryLayer {
 
         // Write a new base image layer at the cutoff point
         let image_layer =
-            ImageLayer::create_from_src(self.conf, timeline, self, end_lsn_inclusive)?;
+            ImageLayer::create_from_src(self.conf, timeline, self, end_lsn_inclusive, batch_fsync)?;
         trace!(
             "freeze: created image layer {} at {}",
             self.seg,

pageserver/src/layered_repository/layer_map.rs

@@ -21,6 +21,8 @@ use std::sync::Arc;
 use zenith_metrics::{register_int_gauge, IntGauge};
 use zenith_utils::lsn::Lsn;
 
+use super::global_layer_map::{LayerId, GLOBAL_LAYER_MAP};
+
 lazy_static! {
     static ref NUM_INMEMORY_LAYERS: IntGauge =
         register_int_gauge!("pageserver_inmemory_layers", "Number of layers in memory")
@@ -68,7 +70,9 @@ impl LayerMap {
     pub fn get_open(&self, tag: &SegmentTag) -> Option<Arc<InMemoryLayer>> {
         let segentry = self.segs.get(tag)?;
 
-        segentry.open.as_ref().map(Arc::clone)
+        segentry
+            .open_layer_id
+            .and_then(|layer_id| GLOBAL_LAYER_MAP.read().unwrap().get(&layer_id))
     }
 
     ///
@@ -77,7 +81,7 @@ impl LayerMap {
     pub fn insert_open(&mut self, layer: Arc<InMemoryLayer>) {
         let segentry = self.segs.entry(layer.get_seg_tag()).or_default();
 
-        segentry.update_open(Arc::clone(&layer));
+        let layer_id = segentry.update_open(Arc::clone(&layer));
 
         let oldest_pending_lsn = layer.get_oldest_pending_lsn();
 
@@ -89,7 +93,7 @@ impl LayerMap {
         // Also add it to the binary heap
         let open_layer_entry = OpenLayerEntry {
             oldest_pending_lsn: layer.get_oldest_pending_lsn(),
-            layer,
+            layer_id,
             generation: self.current_generation,
         };
         self.open_layers.push(open_layer_entry);
@@ -97,24 +101,35 @@ impl LayerMap {
         NUM_INMEMORY_LAYERS.inc();
     }
 
-    /// Remove the oldest in-memory layer
-    pub fn pop_oldest_open(&mut self) {
-        // Pop it from the binary heap
-        let oldest_entry = self.open_layers.pop().unwrap();
-        let segtag = oldest_entry.layer.get_seg_tag();
+    /// Remove an open in-memory layer
+    pub fn remove_open(&mut self, layer_id: LayerId) {
+        // Note: we don't try to remove the entry from the binary heap.
+        // It will be removed lazily by peek_oldest_open() when it's made it to
+        // the top of the heap.
 
-        // Also remove it from the SegEntry of this segment
-        let mut segentry = self.segs.get_mut(&segtag).unwrap();
-        if Arc::ptr_eq(segentry.open.as_ref().unwrap(), &oldest_entry.layer) {
-            segentry.open = None;
-        } else {
-            // We could have already updated segentry.open for
-            // dropped (non-writeable) layer. This is fine.
-            assert!(!oldest_entry.layer.is_writeable());
-            assert!(oldest_entry.layer.is_dropped());
+        let layer_opt = {
+            let mut global_map = GLOBAL_LAYER_MAP.write().unwrap();
+            let layer_opt = global_map.get(&layer_id);
+            global_map.remove(&layer_id);
+            // TODO it's bad that a ref can still exist after being evicted from cache
+            layer_opt
+        };
+
+        if let Some(layer) = layer_opt {
+            let mut segentry = self.segs.get_mut(&layer.get_seg_tag()).unwrap();
+
+            if segentry.open_layer_id == Some(layer_id) {
+                // Also remove it from the SegEntry of this segment
+                segentry.open_layer_id = None;
+            } else {
+                // We could have already updated segentry.open for
+                // dropped (non-writeable) layer. This is fine.
+                assert!(!layer.is_writeable());
+                assert!(layer.is_dropped());
+            }
+
+            NUM_INMEMORY_LAYERS.dec();
         }
-
-        NUM_INMEMORY_LAYERS.dec();
     }
 
     ///
@@ -199,10 +214,17 @@ impl LayerMap {
     }
 
     /// Return the oldest in-memory layer, along with its generation number.
-    pub fn peek_oldest_open(&self) -> Option<(Arc<InMemoryLayer>, u64)> {
-        self.open_layers
-            .peek()
-            .map(|oldest_entry| (Arc::clone(&oldest_entry.layer), oldest_entry.generation))
+    pub fn peek_oldest_open(&mut self) -> Option<(LayerId, Arc<InMemoryLayer>, u64)> {
+        let global_map = GLOBAL_LAYER_MAP.read().unwrap();
+
+        while let Some(oldest_entry) = self.open_layers.peek() {
+            if let Some(layer) = global_map.get(&oldest_entry.layer_id) {
+                return Some((oldest_entry.layer_id, layer, oldest_entry.generation));
+            } else {
+                self.open_layers.pop();
+            }
+        }
+        None
     }
 
     /// Increment the generation number used to stamp open in-memory layers. Layers
@@ -225,8 +247,12 @@ impl LayerMap {
     pub fn dump(&self) -> Result<()> {
         println!("Begin dump LayerMap");
         for (seg, segentry) in self.segs.iter() {
-            if let Some(open) = &segentry.open {
-                open.dump()?;
+            if let Some(open) = &segentry.open_layer_id {
+                if let Some(layer) = GLOBAL_LAYER_MAP.read().unwrap().get(open) {
+                    layer.dump()?;
+                } else {
+                    println!("layer not found in global map");
+                }
             }
 
             for layer in segentry.historic.iter() {
@@ -259,7 +285,7 @@ impl IntervalItem for dyn Layer {
 /// IntervalTree.
 #[derive(Default)]
 struct SegEntry {
-    open: Option<Arc<InMemoryLayer>>,
+    open_layer_id: Option<LayerId>,
     historic: IntervalTree<dyn Layer>,
 }
 
@@ -275,10 +301,10 @@ impl SegEntry {
     }
 
     pub fn get(&self, lsn: Lsn) -> Option<Arc<dyn Layer>> {
-        if let Some(open) = &self.open {
-            if open.get_start_lsn() <= lsn {
-                let x: Arc<dyn Layer> = Arc::clone(open) as _;
-                return Some(x);
+        if let Some(open_layer_id) = &self.open_layer_id {
+            let open_layer = GLOBAL_LAYER_MAP.read().unwrap().get(open_layer_id)?;
+            if open_layer.get_start_lsn() <= lsn {
+                return Some(open_layer);
             }
         }
 
@@ -297,11 +323,16 @@ impl SegEntry {
     // Set new open layer for a SegEntry.
     // It's ok to rewrite previous open layer,
     // but only if it is not writeable anymore.
-    pub fn update_open(&mut self, layer: Arc<InMemoryLayer>) {
-        if let Some(prev_open) = &self.open {
-            assert!(!prev_open.is_writeable());
+    pub fn update_open(&mut self, layer: Arc<InMemoryLayer>) -> LayerId {
+        if let Some(prev_open_layer_id) = &self.open_layer_id {
+            if let Some(prev_open_layer) = GLOBAL_LAYER_MAP.read().unwrap().get(prev_open_layer_id)
+            {
+                assert!(!prev_open_layer.is_writeable());
+            }
         }
-        self.open = Some(layer);
+        let open_layer_id = GLOBAL_LAYER_MAP.write().unwrap().insert(layer);
+        self.open_layer_id = Some(open_layer_id);
+        open_layer_id
     }
 
     pub fn insert_historic(&mut self, layer: Arc<dyn Layer>) {
@@ -316,9 +347,9 @@ impl SegEntry {
 /// recently-added entries (i.e after last call to increment_generation()) from older
 /// entries with the same 'oldest_pending_lsn'.
 struct OpenLayerEntry {
-    pub oldest_pending_lsn: Lsn, // copy of layer.get_oldest_pending_lsn()
-    pub generation: u64,
-    pub layer: Arc<InMemoryLayer>,
+    oldest_pending_lsn: Lsn, // copy of layer.get_oldest_pending_lsn()
+    generation: u64,
+    layer_id: LayerId,
 }
 impl Ord for OpenLayerEntry {
     fn cmp(&self, other: &Self) -> Ordering {
@@ -426,10 +457,10 @@ mod tests {
         // A helper function (closure) to pop the next oldest open entry from the layer map,
         // and assert that it is what we'd expect
         let mut assert_pop_layer = |expected_segno: u32, expected_generation: u64| {
-            let (l, generation) = layers.peek_oldest_open().unwrap();
+            let (layer_id, l, generation) = layers.peek_oldest_open().unwrap();
             assert!(l.get_seg_tag().segno == expected_segno);
             assert!(generation == expected_generation);
-            layers.pop_oldest_open();
+            layers.remove_open(layer_id);
         };
 
         assert_pop_layer(0, gen1); // 0x100

pageserver/src/layered_repository/metadata.rs

@@ -0,0 +1,202 @@
+//! Every image of a certain timeline from [`crate::layered_repository::LayeredRepository`]
+//! has a metadata that needs to be stored persistently.
+//!
+//! Later, the file gets is used in [`crate::relish_storage::storage_sync`] as a part of
+//! external storage import and export operations.
+//!
+//! The module contains all structs and related helper methods related to timeline metadata.
+
+use std::{convert::TryInto, path::PathBuf};
+
+use anyhow::ensure;
+use zenith_utils::{
+    bin_ser::BeSer,
+    lsn::Lsn,
+    zid::{ZTenantId, ZTimelineId},
+};
+
+use crate::{
+    layered_repository::{METADATA_CHECKSUM_SIZE, METADATA_MAX_DATA_SIZE, METADATA_MAX_SAFE_SIZE},
+    PageServerConf,
+};
+
+/// The name of the metadata file pageserver creates per timeline.
+pub const METADATA_FILE_NAME: &str = "metadata";
+
+/// Metadata stored on disk for each timeline
+///
+/// The fields correspond to the values we hold in memory, in LayeredTimeline.
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
+pub struct TimelineMetadata {
+    disk_consistent_lsn: Lsn,
+    // This is only set if we know it. We track it in memory when the page
+    // server is running, but we only track the value corresponding to
+    // 'last_record_lsn', not 'disk_consistent_lsn' which can lag behind by a
+    // lot. We only store it in the metadata file when we flush *all* the
+    // in-memory data so that 'last_record_lsn' is the same as
+    // 'disk_consistent_lsn'.  That's OK, because after page server restart, as
+    // soon as we reprocess at least one record, we will have a valid
+    // 'prev_record_lsn' value in memory again. This is only really needed when
+    // doing a clean shutdown, so that there is no more WAL beyond
+    // 'disk_consistent_lsn'
+    prev_record_lsn: Option<Lsn>,
+    ancestor_timeline: Option<ZTimelineId>,
+    ancestor_lsn: Lsn,
+}
+
+/// Points to a place in pageserver's local directory,
+/// where certain timeline's metadata file should be located.
+pub fn metadata_path(
+    conf: &'static PageServerConf,
+    timelineid: ZTimelineId,
+    tenantid: ZTenantId,
+) -> PathBuf {
+    conf.timeline_path(&timelineid, &tenantid)
+        .join(METADATA_FILE_NAME)
+}
+
+impl TimelineMetadata {
+    pub fn new(
+        disk_consistent_lsn: Lsn,
+        prev_record_lsn: Option<Lsn>,
+        ancestor_timeline: Option<ZTimelineId>,
+        ancestor_lsn: Lsn,
+    ) -> Self {
+        Self {
+            disk_consistent_lsn,
+            prev_record_lsn,
+            ancestor_timeline,
+            ancestor_lsn,
+        }
+    }
+
+    pub fn from_bytes(metadata_bytes: &[u8]) -> anyhow::Result<Self> {
+        ensure!(
+            metadata_bytes.len() == METADATA_MAX_SAFE_SIZE,
+            "metadata bytes size is wrong"
+        );
+
+        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,
+            "metadata checksum mismatch"
+        );
+
+        let data = TimelineMetadata::from(serialize::DeTimelineMetadata::des_prefix(data)?);
+        assert!(data.disk_consistent_lsn.is_aligned());
+
+        Ok(data)
+    }
+
+    pub fn to_bytes(&self) -> anyhow::Result<Vec<u8>> {
+        let serializeable_metadata = serialize::SeTimelineMetadata::from(self);
+        let mut metadata_bytes = serialize::SeTimelineMetadata::ser(&serializeable_metadata)?;
+        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));
+        Ok(metadata_bytes)
+    }
+
+    /// [`Lsn`] that corresponds to the corresponding timeline directory
+    /// contents, stored locally in the pageserver workdir.
+    pub fn disk_consistent_lsn(&self) -> Lsn {
+        self.disk_consistent_lsn
+    }
+
+    pub fn prev_record_lsn(&self) -> Option<Lsn> {
+        self.prev_record_lsn
+    }
+
+    pub fn ancestor_timeline(&self) -> Option<ZTimelineId> {
+        self.ancestor_timeline
+    }
+
+    pub fn ancestor_lsn(&self) -> Lsn {
+        self.ancestor_lsn
+    }
+}
+
+/// This module is for direct conversion of metadata to bytes and back.
+/// For a certain metadata, besides the conversion a few verification steps has to
+/// be done, so all serde derives are hidden from the user, to avoid accidental
+/// verification-less metadata creation.
+mod serialize {
+    use serde::{Deserialize, Serialize};
+    use zenith_utils::{lsn::Lsn, zid::ZTimelineId};
+
+    use super::TimelineMetadata;
+
+    #[derive(Serialize)]
+    pub(super) struct SeTimelineMetadata<'a> {
+        disk_consistent_lsn: &'a Lsn,
+        prev_record_lsn: &'a Option<Lsn>,
+        ancestor_timeline: &'a Option<ZTimelineId>,
+        ancestor_lsn: &'a Lsn,
+    }
+
+    impl<'a> From<&'a TimelineMetadata> for SeTimelineMetadata<'a> {
+        fn from(other: &'a TimelineMetadata) -> Self {
+            Self {
+                disk_consistent_lsn: &other.disk_consistent_lsn,
+                prev_record_lsn: &other.prev_record_lsn,
+                ancestor_timeline: &other.ancestor_timeline,
+                ancestor_lsn: &other.ancestor_lsn,
+            }
+        }
+    }
+
+    #[derive(Deserialize)]
+    pub(super) struct DeTimelineMetadata {
+        disk_consistent_lsn: Lsn,
+        prev_record_lsn: Option<Lsn>,
+        ancestor_timeline: Option<ZTimelineId>,
+        ancestor_lsn: Lsn,
+    }
+
+    impl From<DeTimelineMetadata> for TimelineMetadata {
+        fn from(other: DeTimelineMetadata) -> Self {
+            Self {
+                disk_consistent_lsn: other.disk_consistent_lsn,
+                prev_record_lsn: other.prev_record_lsn,
+                ancestor_timeline: other.ancestor_timeline,
+                ancestor_lsn: other.ancestor_lsn,
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::repository::repo_harness::TIMELINE_ID;
+
+    use super::*;
+
+    #[test]
+    fn metadata_serializes_correctly() {
+        let original_metadata = TimelineMetadata {
+            disk_consistent_lsn: Lsn(0x200),
+            prev_record_lsn: Some(Lsn(0x100)),
+            ancestor_timeline: Some(TIMELINE_ID),
+            ancestor_lsn: Lsn(0),
+        };
+
+        let metadata_bytes = original_metadata
+            .to_bytes()
+            .expect("Should serialize correct metadata to bytes");
+
+        let deserialized_metadata = TimelineMetadata::from_bytes(&metadata_bytes)
+            .expect("Should deserialize its own bytes");
+
+        assert_eq!(
+            deserialized_metadata, original_metadata,
+            "Metadata that was serialized to bytes and deserialized back should not change"
+        );
+    }
+}

pageserver/src/layered_repository/storage_layer.rs

@@ -4,7 +4,7 @@
 
 use crate::relish::RelishTag;
 use crate::repository::WALRecord;
-use crate::ZTimelineId;
+use crate::{ZTenantId, ZTimelineId};
 use anyhow::Result;
 use bytes::Bytes;
 use serde::{Deserialize, Serialize};
@@ -104,6 +104,8 @@ pub enum PageReconstructResult {
 /// in-memory and on-disk layers.
 ///
 pub trait Layer: Send + Sync {
+    fn get_tenant_id(&self) -> ZTenantId;
+
     /// Identify the timeline this relish belongs to
     fn get_timeline_id(&self) -> ZTimelineId;
 

pageserver/src/lib.rs

@@ -18,6 +18,7 @@ pub mod relish_storage;
 pub mod repository;
 pub mod restore_local_repo;
 pub mod tenant_mgr;
+pub mod tenant_threads;
 pub mod waldecoder;
 pub mod walreceiver;
 pub mod walredo;
@@ -42,6 +43,8 @@ pub mod defaults {
 
     pub const DEFAULT_SUPERUSER: &str = "zenith_admin";
     pub const DEFAULT_RELISH_STORAGE_MAX_CONCURRENT_SYNC_LIMITS: usize = 100;
+
+    pub const DEFAULT_OPEN_MEM_LIMIT: usize = 128 * 1024 * 1024;
 }
 
 lazy_static! {
@@ -70,6 +73,8 @@ pub struct PageServerConf {
     pub gc_period: Duration,
     pub superuser: String,
 
+    pub open_mem_limit: usize,
+
     // Repository directory, relative to current working directory.
     // Normally, the page server changes the current working directory
     // to the repository, and 'workdir' is always '.'. But we don't do
@@ -152,6 +157,7 @@ impl PageServerConf {
             checkpoint_period: Duration::from_secs(10),
             gc_horizon: defaults::DEFAULT_GC_HORIZON,
             gc_period: Duration::from_secs(10),
+            open_mem_limit: defaults::DEFAULT_OPEN_MEM_LIMIT,
             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(),
@@ -164,6 +170,15 @@ impl PageServerConf {
     }
 }
 
+/// Config for the Repository checkpointer
+#[derive(Debug, Clone, Copy)]
+pub enum CheckpointConfig {
+    // Flush in-memory data that is older than this
+    Distance(u64),
+    // Flush all in-memory data
+    Forced,
+}
+
 /// External relish storage configuration, enough for creating a client for that storage.
 #[derive(Debug, Clone)]
 pub struct RelishStorageConfig {

pageserver/src/page_service.rs

@@ -630,7 +630,9 @@ impl postgres_backend::Handler for PageServerHandler {
 
             let tenantid = ZTenantId::from_str(caps.get(1).unwrap().as_str())?;
 
-            let branches = crate::branches::get_branches(self.conf, &tenantid)?;
+            // since these handlers for tenant/branch commands are deprecated (in favor of http based ones)
+            // just use false in place of include non incremental logical size
+            let branches = crate::branches::get_branches(self.conf, &tenantid, false)?;
             let branches_buf = serde_json::to_vec(&branches)?;
 
             pgb.write_message_noflush(&SINGLE_COL_ROWDESC)?

pageserver/src/relish_storage/local_fs.rs

@@ -16,8 +16,9 @@ use anyhow::{bail, Context};
 use tokio::{fs, io};
 use tracing::*;
 
+use crate::layered_repository::metadata::METADATA_FILE_NAME;
+
 use super::{parse_ids_from_path, strip_path_prefix, RelishStorage, RemoteRelishInfo};
-use crate::layered_repository::METADATA_FILE_NAME;
 
 pub struct LocalFs {
     pageserver_workdir: &'static Path,
@@ -214,6 +215,7 @@ async fn create_target_directory(target_file_path: &Path) -> anyhow::Result<()>
 #[cfg(test)]
 mod pure_tests {
     use crate::{
+        layered_repository::metadata::METADATA_FILE_NAME,
         relish_storage::test_utils::{
             custom_tenant_id_path, custom_timeline_id_path, relative_timeline_path,
         },

pageserver/src/relish_storage/rust_s3.rs

@@ -11,7 +11,7 @@ use anyhow::Context;
 use s3::{bucket::Bucket, creds::Credentials, region::Region};
 
 use crate::{
-    layered_repository::METADATA_FILE_NAME,
+    layered_repository::metadata::METADATA_FILE_NAME,
     relish_storage::{parse_ids_from_path, strip_path_prefix, RelishStorage, RemoteRelishInfo},
     S3Config,
 };

pageserver/src/relish_storage/storage_sync.rs

@@ -75,7 +75,7 @@ use tracing::*;
 
 use super::{RelishStorage, RemoteRelishInfo};
 use crate::{
-    layered_repository::{metadata_path, TimelineMetadata},
+    layered_repository::metadata::{metadata_path, TimelineMetadata},
     tenant_mgr::register_relish_download,
     PageServerConf,
 };
@@ -151,7 +151,9 @@ struct RemoteTimeline {
 
 impl RemoteTimeline {
     fn disk_consistent_lsn(&self) -> Option<Lsn> {
-        self.metadata.as_ref().map(|meta| meta.disk_consistent_lsn)
+        self.metadata
+            .as_ref()
+            .map(|meta| meta.disk_consistent_lsn())
     }
 }
 
@@ -333,7 +335,7 @@ fn latest_timelines(
         if latest_timeline_id != &remote_timeline_id
             && timeline_metadata
                 .as_ref()
-                .map(|metadata| metadata.disk_consistent_lsn)
+                .map(|metadata| metadata.disk_consistent_lsn())
                 < remote_timeline_data.disk_consistent_lsn()
         {
             *latest_timeline_id = remote_timeline_id;
@@ -518,8 +520,8 @@ async fn upload_timeline<'a, P, S: 'static + RelishStorage<RelishStoragePath = P
         match &uploaded_timeline_files.metadata {
             None => debug!("Partially uploaded timeline found, downloading missing files only"),
             Some(remote_metadata) => {
-                let new_lsn = new_upload.metadata.disk_consistent_lsn;
-                let remote_lsn = remote_metadata.disk_consistent_lsn;
+                let new_lsn = new_upload.metadata.disk_consistent_lsn();
+                let remote_lsn = remote_metadata.disk_consistent_lsn();
                 match new_lsn.cmp(&remote_lsn) {
                     Ordering::Equal | Ordering::Less => {
                         warn!(
@@ -903,7 +905,7 @@ mod tests {
 
         let new_upload_metadata = dummy_metadata(Lsn(0x20));
         assert!(
-            new_upload_metadata.disk_consistent_lsn < first_upload_metadata.disk_consistent_lsn
+            new_upload_metadata.disk_consistent_lsn() < first_upload_metadata.disk_consistent_lsn()
         );
         let new_upload =
             create_local_timeline(&repo_harness, TIMELINE_ID, &["b", "c"], new_upload_metadata)?;
@@ -927,7 +929,8 @@ mod tests {
         )?;
         let second_paths = second_timeline.layers.clone();
         assert!(
-            first_upload_metadata.disk_consistent_lsn < second_upload_metadata.disk_consistent_lsn
+            first_upload_metadata.disk_consistent_lsn()
+                < second_upload_metadata.disk_consistent_lsn()
         );
         ensure_correct_timeline_upload(
             &repo_harness,
@@ -955,7 +958,8 @@ mod tests {
 
         let third_upload_metadata = dummy_metadata(Lsn(0x50));
         assert!(
-            second_upload_metadata.disk_consistent_lsn < third_upload_metadata.disk_consistent_lsn
+            second_upload_metadata.disk_consistent_lsn()
+                < third_upload_metadata.disk_consistent_lsn()
         );
         let third_timeline = create_local_timeline(
             &repo_harness,
@@ -1249,7 +1253,7 @@ mod tests {
         while let Some(task) = queue_accessor.pop() {
             let task_lsn = match &task {
                 SyncTask::Upload(LocalTimeline { metadata, .. }) => {
-                    Some(metadata.disk_consistent_lsn)
+                    Some(metadata.disk_consistent_lsn())
                 }
                 SyncTask::UrgentDownload(remote_timeline) | SyncTask::Download(remote_timeline) => {
                     remote_timeline.disk_consistent_lsn()
@@ -1257,8 +1261,8 @@ mod tests {
             };
 
             if let Some(task_lsn) = task_lsn {
-                if task_lsn == smaller_lsn_metadata.disk_consistent_lsn
-                    || task_lsn == bigger_lsn_metadata.disk_consistent_lsn
+                if task_lsn == smaller_lsn_metadata.disk_consistent_lsn()
+                    || task_lsn == bigger_lsn_metadata.disk_consistent_lsn()
                 {
                     ordered_tasks.push(task);
                 }
@@ -1549,11 +1553,6 @@ mod tests {
     }
 
     fn dummy_metadata(disk_consistent_lsn: Lsn) -> TimelineMetadata {
-        TimelineMetadata {
-            disk_consistent_lsn,
-            prev_record_lsn: None,
-            ancestor_timeline: None,
-            ancestor_lsn: Lsn(0),
-        }
+        TimelineMetadata::new(disk_consistent_lsn, None, None, Lsn(0))
     }
 }

pageserver/src/repository.rs

@@ -1,4 +1,5 @@
 use crate::relish::*;
+use crate::CheckpointConfig;
 use anyhow::Result;
 use bytes::{Buf, BufMut, Bytes, BytesMut};
 use serde::{Deserialize, Serialize};
@@ -24,9 +25,9 @@ pub trait Repository: Send + Sync {
     /// Branch a timeline
     fn branch_timeline(&self, src: ZTimelineId, dst: ZTimelineId, start_lsn: Lsn) -> Result<()>;
 
-    /// perform one garbage collection iteration.
-    /// garbage collection is periodically performed by gc thread,
-    /// but it can be explicitly requested through page server api.
+    /// perform one garbage collection iteration, removing old data files from disk.
+    /// this funtion is periodically called by gc thread.
+    /// also it can be explicitly requested through page server api 'do_gc' command.
     ///
     /// 'timelineid' specifies the timeline to GC, or None for all.
     /// `horizon` specifies delta from last lsn to preserve all object versions (pitr interval).
@@ -39,6 +40,10 @@ pub trait Repository: Send + Sync {
         horizon: u64,
         checkpoint_before_gc: bool,
     ) -> Result<GcResult>;
+
+    /// perform one checkpoint iteration, flushing in-memory data on disk.
+    /// this function is periodically called by checkponter thread.
+    fn checkpoint_iteration(&self, cconf: CheckpointConfig) -> Result<()>;
 }
 
 ///
@@ -144,7 +149,7 @@ pub trait Timeline: Send + Sync {
     ///
     /// NOTE: This has nothing to do with checkpoint in PostgreSQL. We don't
     /// know anything about them here in the repository.
-    fn checkpoint(&self) -> Result<()>;
+    fn checkpoint(&self, cconf: CheckpointConfig) -> Result<()>;
 
     /// Retrieve current logical size of the timeline
     ///
@@ -155,6 +160,9 @@ pub trait Timeline: Send + Sync {
     /// Does the same as get_current_logical_size but counted on demand.
     /// Used in tests to ensure thet incremental and non incremental variants match.
     fn get_current_logical_size_non_incremental(&self, lsn: Lsn) -> Result<usize>;
+
+    /// An escape hatch to allow "casting" a generic Timeline to LayeredTimeline.
+    fn upgrade_to_layered_timeline(&self) -> &crate::layered_repository::LayeredTimeline;
 }
 
 /// Various functions to mutate the timeline.
@@ -318,9 +326,10 @@ pub mod repo_harness {
 #[allow(clippy::bool_assert_comparison)]
 #[cfg(test)]
 mod tests {
+    use crate::layered_repository::metadata::METADATA_FILE_NAME;
+
     use super::repo_harness::*;
     use super::*;
-    use crate::layered_repository::METADATA_FILE_NAME;
     use postgres_ffi::{pg_constants, xlog_utils::SIZEOF_CHECKPOINT};
 
     /// Arbitrary relation tag, for testing.
@@ -714,7 +723,7 @@ mod tests {
             .contains(&TESTREL_A));
 
         // Run checkpoint and garbage collection and check that it's still not visible
-        newtline.checkpoint()?;
+        newtline.checkpoint(CheckpointConfig::Forced)?;
         repo.gc_iteration(Some(NEW_TIMELINE_ID), 0, true)?;
 
         assert!(!newtline

pageserver/src/tenant_mgr.rs

@@ -4,6 +4,7 @@
 use crate::branches;
 use crate::layered_repository::LayeredRepository;
 use crate::repository::{Repository, Timeline};
+use crate::tenant_threads;
 use crate::walredo::PostgresRedoManager;
 use crate::PageServerConf;
 use anyhow::{anyhow, bail, Context, Result};
@@ -16,7 +17,6 @@ use std::fs;
 use std::str::FromStr;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex, MutexGuard};
-use std::thread::JoinHandle;
 use zenith_utils::zid::{ZTenantId, ZTimelineId};
 
 lazy_static! {
@@ -28,7 +28,7 @@ struct Tenant {
     repo: Option<Arc<dyn Repository>>,
 }
 
-#[derive(Debug, Serialize, Deserialize, Clone, Copy)]
+#[derive(Debug, Serialize, Deserialize, Clone, Copy, PartialEq, Eq)]
 pub enum TenantState {
     // This tenant only exists in cloud storage. It cannot be accessed.
     CloudOnly,
@@ -41,10 +41,12 @@ pub enum TenantState {
     // This tenant exists on local disk, and the layer map has been loaded into memory.
     // The local disk might have some newer files that don't exist in cloud storage yet.
     Active,
+    // Tenant is active, but there is no walreceiver connection.
+    Idle,
     // This tenant exists on local disk, and the layer map has been loaded into memory.
     // The local disk might have some newer files that don't exist in cloud storage yet.
     // The tenant cannot be accessed anymore for any reason, but graceful shutdown.
-    //Stopping,
+    Stopping,
 }
 
 impl fmt::Display for TenantState {
@@ -53,6 +55,8 @@ impl fmt::Display for TenantState {
             TenantState::CloudOnly => f.write_str("CloudOnly"),
             TenantState::Downloading => f.write_str("Downloading"),
             TenantState::Active => f.write_str("Active"),
+            TenantState::Idle => f.write_str("Idle"),
+            TenantState::Stopping => f.write_str("Stopping"),
         }
     }
 }
@@ -61,18 +65,6 @@ fn access_tenants() -> MutexGuard<'static, HashMap<ZTenantId, Tenant>> {
     TENANTS.lock().unwrap()
 }
 
-struct TenantHandleEntry {
-    checkpointer_handle: Option<JoinHandle<()>>,
-    gc_handle: Option<JoinHandle<()>>,
-}
-
-// Logically these handles belong to Repository,
-// but it's just simpler to store them separately
-lazy_static! {
-    static ref TENANT_HANDLES: Mutex<HashMap<ZTenantId, TenantHandleEntry>> =
-        Mutex::new(HashMap::new());
-}
-
 static SHUTDOWN_REQUESTED: AtomicBool = AtomicBool::new(false);
 
 pub fn init(conf: &'static PageServerConf) {
@@ -106,21 +98,10 @@ fn init_repo(conf: &'static PageServerConf, tenant_id: ZTenantId) {
         true,
     ));
 
-    let checkpointer_handle = LayeredRepository::launch_checkpointer_thread(conf, repo.clone());
-    let gc_handle = LayeredRepository::launch_gc_thread(conf, repo.clone());
-
-    let mut handles = TENANT_HANDLES.lock().unwrap();
-    let h = TenantHandleEntry {
-        checkpointer_handle: Some(checkpointer_handle),
-        gc_handle: Some(gc_handle),
-    };
-
-    handles.insert(tenant_id, h);
-
     let mut m = access_tenants();
     let tenant = m.get_mut(&tenant_id).unwrap();
     tenant.repo = Some(repo);
-    tenant.state = TenantState::Active;
+    tenant.state = TenantState::Idle;
 }
 
 pub fn register_relish_download(
@@ -142,10 +123,14 @@ pub fn register_relish_download(
         });
         tenant.state = TenantState::Downloading;
         match &tenant.repo {
-            Some(repo) => init_timeline(repo.as_ref(), timeline_id),
+            Some(repo) => {
+                init_timeline(repo.as_ref(), timeline_id);
+                tenant.state = TenantState::Idle;
+                return;
+            }
             None => log::warn!("Initialize new repo"),
         }
-        tenant.state = TenantState::Active;
+        tenant.state = TenantState::Idle;
     }
 
     // init repo updates Tenant state
@@ -166,27 +151,23 @@ pub fn shutdown_requested() -> bool {
     SHUTDOWN_REQUESTED.load(Ordering::Relaxed)
 }
 
-pub fn stop_tenant_threads(tenantid: ZTenantId) {
-    let mut handles = TENANT_HANDLES.lock().unwrap();
-    if let Some(h) = handles.get_mut(&tenantid) {
-        h.checkpointer_handle.take().map(JoinHandle::join);
-        debug!("checkpointer for tenant {} has stopped", tenantid);
-        h.gc_handle.take().map(JoinHandle::join);
-        debug!("gc for tenant {} has stopped", tenantid);
-    }
-}
-
 pub fn shutdown_all_tenants() -> Result<()> {
     SHUTDOWN_REQUESTED.swap(true, Ordering::Relaxed);
 
     let tenantids = list_tenantids()?;
+
+    for tenantid in &tenantids {
+        set_tenant_state(*tenantid, TenantState::Stopping)?;
+    }
+
     for tenantid in tenantids {
-        stop_tenant_threads(tenantid);
+        // Wait for checkpointer and GC to finish their job
+        tenant_threads::wait_for_tenant_threads_to_stop(tenantid);
+
         let repo = get_repository_for_tenant(tenantid)?;
         debug!("shutdown tenant {}", tenantid);
         repo.shutdown()?;
     }
-
     Ok(())
 }
 
@@ -213,11 +194,38 @@ pub fn create_repository_for_tenant(
     let mut m = access_tenants();
     let tenant = m.get_mut(&tenantid).unwrap();
     tenant.repo = Some(repo);
-    tenant.state = TenantState::Active;
+    tenant.state = TenantState::Idle;
 
     Ok(())
 }
 
+// If tenant is not found in the repository, return CloudOnly state
+pub fn get_tenant_state(tenantid: ZTenantId) -> TenantState {
+    let m = access_tenants();
+    match m.get(&tenantid) {
+        Some(tenant) => tenant.state,
+        None => TenantState::CloudOnly,
+    }
+}
+
+pub fn set_tenant_state(tenantid: ZTenantId, newstate: TenantState) -> Result<TenantState> {
+    let mut m = access_tenants();
+    let tenant = m.get_mut(&tenantid);
+
+    match tenant {
+        Some(tenant) => {
+            if newstate == TenantState::Idle && tenant.state != TenantState::Active {
+                // Only Active tenant can become Idle
+                return Ok(tenant.state);
+            }
+            info!("set_tenant_state: {} -> {}", tenant.state, newstate);
+            tenant.state = newstate;
+            Ok(tenant.state)
+        }
+        None => bail!("Tenant not found for tenant {}", tenantid),
+    }
+}
+
 pub fn get_repository_for_tenant(tenantid: ZTenantId) -> Result<Arc<dyn Repository>> {
     let m = access_tenants();
     let tenant = m

pageserver/src/tenant_threads.rs

@@ -0,0 +1,149 @@
+//! This module contains functions to serve per-tenant background processes,
+//! such as checkpointer and GC
+use crate::tenant_mgr;
+use crate::tenant_mgr::TenantState;
+use crate::CheckpointConfig;
+use crate::PageServerConf;
+use anyhow::Result;
+use lazy_static::lazy_static;
+use std::collections::HashMap;
+use std::sync::Mutex;
+use std::thread::JoinHandle;
+use std::time::Duration;
+use tracing::*;
+use zenith_metrics::{register_int_gauge_vec, IntGaugeVec};
+use zenith_utils::zid::ZTenantId;
+
+struct TenantHandleEntry {
+    checkpointer_handle: Option<JoinHandle<()>>,
+    gc_handle: Option<JoinHandle<()>>,
+}
+
+// Preserve handles to wait for thread completion
+// at shutdown
+lazy_static! {
+    static ref TENANT_HANDLES: Mutex<HashMap<ZTenantId, TenantHandleEntry>> =
+        Mutex::new(HashMap::new());
+}
+
+lazy_static! {
+    static ref TENANT_THREADS_COUNT: IntGaugeVec = register_int_gauge_vec!(
+        "tenant_threads_count",
+        "Number of live tenant threads",
+        &["tenant_thread_type"]
+    )
+    .expect("failed to define a metric");
+}
+
+// Launch checkpointer and GC for the tenant.
+// It's possible that the threads are running already,
+// if so, just don't spawn new ones.
+pub fn start_tenant_threads(conf: &'static PageServerConf, tenantid: ZTenantId) {
+    let mut handles = TENANT_HANDLES.lock().unwrap();
+    let h = handles
+        .entry(tenantid)
+        .or_insert_with(|| TenantHandleEntry {
+            checkpointer_handle: None,
+            gc_handle: None,
+        });
+
+    if h.checkpointer_handle.is_none() {
+        h.checkpointer_handle = std::thread::Builder::new()
+            .name("Checkpointer thread".into())
+            .spawn(move || {
+                checkpoint_loop(tenantid, conf).expect("Checkpointer thread died");
+            })
+            .ok();
+    }
+
+    if h.gc_handle.is_none() {
+        h.gc_handle = std::thread::Builder::new()
+            .name("GC thread".into())
+            .spawn(move || {
+                gc_loop(tenantid, conf).expect("GC thread died");
+            })
+            .ok();
+    }
+}
+
+pub fn wait_for_tenant_threads_to_stop(tenantid: ZTenantId) {
+    let mut handles = TENANT_HANDLES.lock().unwrap();
+    if let Some(h) = handles.get_mut(&tenantid) {
+        h.checkpointer_handle.take().map(JoinHandle::join);
+        trace!("checkpointer for tenant {} has stopped", tenantid);
+        h.gc_handle.take().map(JoinHandle::join);
+        trace!("gc for tenant {} has stopped", tenantid);
+    }
+    handles.remove(&tenantid);
+}
+
+///
+/// Checkpointer thread's main loop
+///
+fn checkpoint_loop(tenantid: ZTenantId, conf: &'static PageServerConf) -> Result<()> {
+    let gauge = TENANT_THREADS_COUNT.with_label_values(&["checkpointer"]);
+    gauge.inc();
+    scopeguard::defer! {
+        gauge.dec();
+    }
+
+    loop {
+        if tenant_mgr::get_tenant_state(tenantid) != TenantState::Active {
+            break;
+        }
+
+        std::thread::sleep(conf.checkpoint_period);
+        trace!("checkpointer thread for tenant {} waking up", tenantid);
+
+        // checkpoint timelines that have accumulated more than CHECKPOINT_DISTANCE
+        // bytes of WAL since last checkpoint.
+        let repo = tenant_mgr::get_repository_for_tenant(tenantid)?;
+        repo.checkpoint_iteration(CheckpointConfig::Distance(conf.checkpoint_distance))?;
+    }
+
+    trace!(
+        "checkpointer thread stopped for tenant {} state is {}",
+        tenantid,
+        tenant_mgr::get_tenant_state(tenantid)
+    );
+    Ok(())
+}
+
+///
+/// GC thread's main loop
+///
+fn gc_loop(tenantid: ZTenantId, conf: &'static PageServerConf) -> Result<()> {
+    let gauge = TENANT_THREADS_COUNT.with_label_values(&["gc"]);
+    gauge.inc();
+    scopeguard::defer! {
+        gauge.dec();
+    }
+
+    loop {
+        if tenant_mgr::get_tenant_state(tenantid) != TenantState::Active {
+            break;
+        }
+
+        trace!("gc thread for tenant {} waking up", tenantid);
+
+        // Garbage collect old files that are not needed for PITR anymore
+        if conf.gc_horizon > 0 {
+            let repo = tenant_mgr::get_repository_for_tenant(tenantid)?;
+            repo.gc_iteration(None, conf.gc_horizon, false).unwrap();
+        }
+
+        // TODO Write it in more adequate way using
+        // condvar.wait_timeout() or something
+        let mut sleep_time = conf.gc_period.as_secs();
+        while sleep_time > 0 && tenant_mgr::get_tenant_state(tenantid) == TenantState::Active {
+            sleep_time -= 1;
+            std::thread::sleep(Duration::from_secs(1));
+        }
+    }
+    trace!(
+        "GC thread stopped for tenant {} state is {}",
+        tenantid,
+        tenant_mgr::get_tenant_state(tenantid)
+    );
+    Ok(())
+}

pageserver/src/walreceiver.rs

@@ -5,9 +5,12 @@
 //!
 //! We keep one WAL receiver active per timeline.
 
+use crate::layered_repository;
 use crate::relish::*;
 use crate::restore_local_repo;
 use crate::tenant_mgr;
+use crate::tenant_mgr::TenantState;
+use crate::tenant_threads;
 use crate::waldecoder::*;
 use crate::PageServerConf;
 use anyhow::{bail, Error, Result};
@@ -38,6 +41,7 @@ use zenith_utils::zid::ZTimelineId;
 struct WalReceiverEntry {
     wal_producer_connstr: String,
     wal_receiver_handle: Option<JoinHandle<()>>,
+    tenantid: ZTenantId,
 }
 
 lazy_static! {
@@ -65,6 +69,23 @@ pub fn stop_wal_receiver(timelineid: ZTimelineId) {
     }
 }
 
+pub fn drop_wal_receiver(timelineid: ZTimelineId, tenantid: ZTenantId) {
+    let mut receivers = WAL_RECEIVERS.lock().unwrap();
+    receivers.remove(&timelineid);
+
+    // Check if it was the last walreceiver of the tenant.
+    // TODO now we store one WalReceiverEntry per timeline,
+    // so this iterator looks a bit strange.
+    for (_timelineid, entry) in receivers.iter() {
+        if entry.tenantid == tenantid {
+            return;
+        }
+    }
+
+    // When last walreceiver of the tenant is gone, change state to Idle
+    tenant_mgr::set_tenant_state(tenantid, TenantState::Idle).unwrap();
+}
+
 // Launch a new WAL receiver, or tell one that's running about change in connection string
 pub fn launch_wal_receiver(
     conf: &'static PageServerConf,
@@ -90,8 +111,13 @@ pub fn launch_wal_receiver(
             let receiver = WalReceiverEntry {
                 wal_producer_connstr: wal_producer_connstr.into(),
                 wal_receiver_handle: Some(wal_receiver_handle),
+                tenantid,
             };
             receivers.insert(timelineid, receiver);
+
+            // Update tenant state and start tenant threads, if they are not running yet.
+            tenant_mgr::set_tenant_state(tenantid, TenantState::Active).unwrap();
+            tenant_threads::start_tenant_threads(conf, tenantid);
         }
     };
 }
@@ -114,11 +140,15 @@ fn thread_main(conf: &'static PageServerConf, timelineid: ZTimelineId, tenantid:
     let _enter = info_span!("WAL receiver", timeline = %timelineid, tenant = %tenantid).entered();
     info!("WAL receiver thread started");
 
+    let mut retry_count = 10;
+
     //
     // Make a connection to the WAL safekeeper, or directly to the primary PostgreSQL server,
     // and start streaming WAL from it. If the connection is lost, keep retrying.
+    // TODO How long should we retry in case of losing connection?
+    // Should we retry at all or we can wait for the next callmemaybe request?
     //
-    while !tenant_mgr::shutdown_requested() {
+    while !tenant_mgr::shutdown_requested() && retry_count > 0 {
         // Look up the current WAL producer address
         let wal_producer_connstr = get_wal_producer_connstr(timelineid);
 
@@ -129,14 +159,24 @@ fn thread_main(conf: &'static PageServerConf, timelineid: ZTimelineId, tenantid:
                 "WAL streaming connection failed ({}), retrying in 1 second",
                 e
             );
+            retry_count -= 1;
             sleep(Duration::from_secs(1));
+        } else {
+            info!(
+                "walreceiver disconnected tenant {}, timelineid {}",
+                tenantid, timelineid
+            );
+            break;
         }
     }
-    debug!("WAL streaming shut down");
+    info!("WAL streaming shut down");
+    // Drop it from list of active WAL_RECEIVERS
+    // so that next callmemaybe request launched a new thread
+    drop_wal_receiver(timelineid, tenantid);
 }
 
 fn walreceiver_main(
-    _conf: &PageServerConf,
+    conf: &PageServerConf,
     timelineid: ZTimelineId,
     wal_producer_connstr: &str,
     tenantid: ZTenantId,
@@ -256,6 +296,9 @@ fn walreceiver_main(
                     caught_up = true;
                 }
 
+                // Release memory if needed
+                layered_repository::evict_layer_if_needed(conf)?;
+
                 Some(endlsn)
             }
 
@@ -300,6 +343,7 @@ fn walreceiver_main(
             break;
         }
     }
+
     Ok(())
 }
 

proxy/src/cplane_api.rs

@@ -15,6 +15,13 @@ pub struct DatabaseInfo {
     pub password: Option<String>,
 }
 
+#[derive(Serialize, Deserialize, Debug)]
+pub struct ProxyAuthResult {
+    pub ready: bool,
+    pub error: Option<String>,
+    pub conn_info: Option<DatabaseInfo>,
+}
+
 impl DatabaseInfo {
     pub fn socket_addr(&self) -> Result<SocketAddr> {
         let host_port = format!("{}:{}", self.host, self.port);
@@ -55,22 +62,25 @@ impl CPlaneApi {
         database: &str,
         md5_response: &[u8],
         salt: &[u8; 4],
-    ) -> Result<DatabaseInfo> {
+        psql_session_id: &str,
+    ) -> Result<ProxyAuthResult> {
         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));
+            .append_pair("salt", &hex::encode(salt))
+            .append_pair("psql_session_id", psql_session_id);
 
         println!("cplane request: {}", url.as_str());
 
         let resp = reqwest::blocking::get(url)?;
 
         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)
+            let auth_info: ProxyAuthResult = serde_json::from_str(resp.text()?.as_str())?;
+            println!("got auth info: #{:?}", auth_info);
+
+            Ok(auth_info)
         } else {
             bail!("Auth failed")
         }

proxy/src/main.rs

@@ -7,7 +7,7 @@
 ///
 use std::{
     collections::HashMap,
-    net::{SocketAddr, TcpListener},
+    net::SocketAddr,
     sync::{mpsc, Arc, Mutex},
     thread,
 };
@@ -17,6 +17,7 @@ use clap::{App, Arg, ArgMatches};
 
 use cplane_api::DatabaseInfo;
 use rustls::{internal::pemfile, NoClientAuth, ProtocolVersion, ServerConfig};
+use zenith_utils::tcp_listener;
 
 mod cplane_api;
 mod mgmt;
@@ -140,10 +141,10 @@ fn main() -> anyhow::Result<()> {
 
     // Check that we can bind to address before further initialization
     println!("Starting proxy on {}", state.conf.proxy_address);
-    let pageserver_listener = TcpListener::bind(state.conf.proxy_address)?;
+    let pageserver_listener = tcp_listener::bind(state.conf.proxy_address)?;
 
     println!("Starting mgmt on {}", state.conf.mgmt_address);
-    let mgmt_listener = TcpListener::bind(state.conf.mgmt_address)?;
+    let mgmt_listener = tcp_listener::bind(state.conf.mgmt_address)?;
 
     let threads = [
         // Spawn a thread to listen for connections. It will spawn further threads

proxy/src/proxy.rs

@@ -75,8 +75,12 @@ pub fn proxy_conn_main(
     // This will set conn.existing_user and we can decide on next actions
     conn.handle_startup()?;
 
+    let mut psql_session_id_buf = [0u8; 8];
+    rand::thread_rng().fill(&mut psql_session_id_buf);
+    conn.psql_session_id = hex::encode(psql_session_id_buf);
+
     // both scenarious here should end up producing database connection string
-    let db_info = if conn.is_existing_user() {
+    let conn_info = if conn.is_existing_user() {
         conn.handle_existing_user()?
     } else {
         conn.handle_new_user()?
@@ -84,7 +88,7 @@ pub fn proxy_conn_main(
 
     // XXX: move that inside handle_new_user/handle_existing_user to be able to
     // report wrong connection error.
-    proxy_pass(conn.pgb, db_info)
+    proxy_pass(conn.pgb, conn_info)
 }
 
 impl ProxyConnection {
@@ -156,6 +160,21 @@ impl ProxyConnection {
         Ok(())
     }
 
+    // Wait for proxy kick form the console with conninfo
+    fn wait_for_conninfo(&mut self) -> anyhow::Result<DatabaseInfo> {
+        let (tx, rx) = channel::<anyhow::Result<DatabaseInfo>>();
+        let _ = self
+            .state
+            .waiters
+            .lock()
+            .unwrap()
+            .insert(self.psql_session_id.clone(), tx);
+
+        // Wait for web console response
+        // TODO: respond with error to client
+        rx.recv()?
+    }
+
     fn handle_existing_user(&mut self) -> anyhow::Result<DatabaseInfo> {
         // ask password
         rand::thread_rng().fill(&mut self.md5_salt);
@@ -182,14 +201,41 @@ impl ProxyConnection {
                 self.database.as_str(),
                 md5_response,
                 &self.md5_salt,
+                &self.psql_session_id,
             ) {
                 Err(e) => {
-                    self.pgb
-                        .write_message(&BeMessage::ErrorResponse(format!("{}", e)))?;
+                    self.pgb.write_message(&BeMessage::ErrorResponse(format!(
+                        "cannot authenticate proxy: {}",
+                        e
+                    )))?;
 
                     bail!("auth failed: {}", e);
                 }
-                Ok(conn_info) => {
+
+                Ok(auth_info) => {
+                    let conn_info = if auth_info.ready {
+                        // Cluster is ready, so just take `conn_info` and respond to the client.
+                        auth_info
+                            .conn_info
+                            .expect("conn_info should be provided with ready cluster")
+                    } else {
+                        match auth_info.error {
+                            Some(e) => {
+                                self.pgb.write_message(&BeMessage::ErrorResponse(format!(
+                                    "cannot authenticate proxy: {}",
+                                    e
+                                )))?;
+
+                                bail!("auth failed: {}", e);
+                            }
+                            None => {
+                                // Cluster exists, but isn't active, await its start and proxy kick
+                                // with `conn_info`.
+                                self.wait_for_conninfo()?
+                            }
+                        }
+                    };
+
                     self.pgb
                         .write_message_noflush(&BeMessage::AuthenticationOk)?;
                     self.pgb
@@ -205,10 +251,6 @@ impl ProxyConnection {
     }
 
     fn handle_new_user(&mut self) -> anyhow::Result<DatabaseInfo> {
-        let mut psql_session_id_buf = [0u8; 8];
-        rand::thread_rng().fill(&mut psql_session_id_buf);
-        self.psql_session_id = hex::encode(psql_session_id_buf);
-
         let hello_message = format!("☀️  Welcome to Zenith!
 
 To proceed with database creation, open the following link:
@@ -227,25 +269,15 @@ databases without opening the browser.
         self.pgb
             .write_message(&BeMessage::NoticeResponse(hello_message))?;
 
-        // await for database creation
-        let (tx, rx) = channel::<anyhow::Result<DatabaseInfo>>();
-        let _ = self
-            .state
-            .waiters
-            .lock()
-            .unwrap()
-            .insert(self.psql_session_id.clone(), tx);
-
-        // Wait for web console response
-        // XXX: respond with error to client
-        let dbinfo = rx.recv()??;
+        // We requested the DB creation from the console. Now wait for conninfo
+        let conn_info = self.wait_for_conninfo()?;
 
         self.pgb.write_message_noflush(&BeMessage::NoticeResponse(
             "Connecting to database.".to_string(),
         ))?;
         self.pgb.write_message(&BeMessage::ReadyForQuery)?;
 
-        Ok(dbinfo)
+        Ok(conn_info)
     }
 }
 

test_runner/fixtures/zenith_fixtures.py

@@ -610,7 +610,9 @@ class ZenithPageserverHttpClient(requests.Session):
         return res_json
 
     def branch_detail(self, tenant_id: uuid.UUID, name: str) -> Dict[Any, Any]:
-        res = self.get(f"http://localhost:{self.port}/v1/branch/{tenant_id.hex}/{name}", )
+        res = self.get(
+            f"http://localhost:{self.port}/v1/branch/{tenant_id.hex}/{name}?include-non-incremental-logical-size=1",
+        )
         res.raise_for_status()
         res_json = res.json()
         assert isinstance(res_json, dict)

walkeeper/src/bin/safekeeper.rs

@@ -7,11 +7,10 @@ use const_format::formatcp;
 use daemonize::Daemonize;
 use log::*;
 use std::env;
-use std::net::TcpListener;
 use std::path::{Path, PathBuf};
 use std::thread;
 use zenith_utils::http::endpoint;
-use zenith_utils::logging;
+use zenith_utils::{logging, tcp_listener};
 
 use walkeeper::defaults::{DEFAULT_HTTP_LISTEN_ADDR, DEFAULT_PG_LISTEN_ADDR};
 use walkeeper::http;
@@ -132,13 +131,13 @@ fn main() -> Result<()> {
 fn start_safekeeper(conf: SafeKeeperConf) -> Result<()> {
     let log_file = logging::init("safekeeper.log", conf.daemonize)?;
 
-    let http_listener = TcpListener::bind(conf.listen_http_addr.clone()).map_err(|e| {
+    let http_listener = tcp_listener::bind(conf.listen_http_addr.clone()).map_err(|e| {
         error!("failed to bind to address {}: {}", conf.listen_http_addr, e);
         e
     })?;
 
     info!("Starting safekeeper on {}", conf.listen_pg_addr);
-    let pg_listener = TcpListener::bind(conf.listen_pg_addr.clone()).map_err(|e| {
+    let pg_listener = tcp_listener::bind(conf.listen_pg_addr.clone()).map_err(|e| {
         error!("failed to bind to address {}: {}", conf.listen_pg_addr, e);
         e
     })?;

walkeeper/src/receive_wal.rs

@@ -133,9 +133,12 @@ impl<'pg> ReceiveWalConn<'pg> {
             // Add far as replication in postgres is initiated by receiver, we should use callme mechanism
             let conf = swh.conf.clone();
             let timelineid = swh.timeline.get().timelineid;
-            thread::spawn(move || {
-                request_callback(conf, timelineid, tenant_id);
-            });
+            let _ = thread::Builder::new()
+                .name("request_callback thread".into())
+                .spawn(move || {
+                    request_callback(conf, timelineid, tenant_id);
+                })
+                .unwrap();
         }
 
         loop {

walkeeper/src/replication.rs

@@ -173,11 +173,14 @@ impl ReplicationConn {
         let bg_timeline = Arc::clone(swh.timeline.get());
         let bg_stream_in = self.stream_in.take().unwrap();
 
-        thread::spawn(move || {
-            if let Err(err) = Self::background_thread(bg_stream_in, bg_timeline) {
-                error!("Replication background thread failed: {}", err);
-            }
-        });
+        let _ = thread::Builder::new()
+            .name("HotStandbyFeedback thread".into())
+            .spawn(move || {
+                if let Err(err) = Self::background_thread(bg_stream_in, bg_timeline) {
+                    error!("Replication background thread failed: {}", err);
+                }
+            })
+            .unwrap();
 
         let (mut start_pos, mut stop_pos) = Self::parse_start_stop(cmd)?;
 

walkeeper/src/safekeeper.rs

@@ -19,7 +19,10 @@ use lazy_static::lazy_static;
 
 use crate::replication::HotStandbyFeedback;
 use postgres_ffi::xlog_utils::MAX_SEND_SIZE;
-use zenith_metrics::{register_gauge_vec, Gauge, GaugeVec};
+use zenith_metrics::{
+    register_gauge_vec, register_histogram_vec, Gauge, GaugeVec, Histogram, HistogramVec,
+    DISK_WRITE_SECONDS_BUCKETS,
+};
 use zenith_utils::bin_ser::LeSer;
 use zenith_utils::lsn::Lsn;
 use zenith_utils::pq_proto::SystemId;
@@ -299,11 +302,27 @@ lazy_static! {
         &["ztli"]
     )
     .expect("Failed to register safekeeper_commit_lsn gauge vec");
+    static ref WRITE_WAL_BYTES: HistogramVec = register_histogram_vec!(
+        "safekeeper_write_wal_bytes",
+        "Bytes written to WAL in a single request, grouped by timeline",
+        &["timeline_id"],
+        vec![1.0, 10.0, 100.0, 1024.0, 8192.0, 128.0 * 1024.0, 1024.0 * 1024.0, 10.0 * 1024.0 * 1024.0]
+    )
+    .expect("Failed to register safekeeper_write_wal_bytes histogram vec");
+    static ref WRITE_WAL_SECONDS: HistogramVec = register_histogram_vec!(
+        "safekeeper_write_wal_seconds",
+        "Seconds spent writing and syncing WAL to a disk in a single request, grouped by timeline",
+        &["timeline_id"],
+        DISK_WRITE_SECONDS_BUCKETS.to_vec()
+    )
+    .expect("Failed to register safekeeper_write_wal_seconds histogram vec");
 }
 
 struct SafeKeeperMetrics {
     flush_lsn: Gauge,
     commit_lsn: Gauge,
+    write_wal_bytes: Histogram,
+    write_wal_seconds: Histogram,
 }
 
 impl SafeKeeperMetrics {
@@ -312,6 +331,8 @@ impl SafeKeeperMetrics {
         SafeKeeperMetrics {
             flush_lsn: FLUSH_LSN_GAUGE.with_label_values(&[&ztli_str]),
             commit_lsn: COMMIT_LSN_GAUGE.with_label_values(&[&ztli_str]),
+            write_wal_bytes: WRITE_WAL_BYTES.with_label_values(&[&ztli_str]),
+            write_wal_seconds: WRITE_WAL_SECONDS.with_label_values(&[&ztli_str]),
         }
     }
 
@@ -319,6 +340,8 @@ impl SafeKeeperMetrics {
         SafeKeeperMetrics {
             flush_lsn: FLUSH_LSN_GAUGE.with_label_values(&["n/a"]),
             commit_lsn: COMMIT_LSN_GAUGE.with_label_values(&["n/a"]),
+            write_wal_bytes: WRITE_WAL_BYTES.with_label_values(&["n/a"]),
+            write_wal_seconds: WRITE_WAL_SECONDS.with_label_values(&["n/a"]),
         }
     }
 }
@@ -472,8 +495,14 @@ where
         // do the job
         let mut last_rec_lsn = Lsn(0);
         if !msg.wal_data.is_empty() {
-            self.storage
-                .write_wal(&self.s.server, msg.h.begin_lsn, &msg.wal_data)?;
+            self.metrics
+                .write_wal_bytes
+                .observe(msg.wal_data.len() as f64);
+            {
+                let _timer = self.metrics.write_wal_seconds.start_timer();
+                self.storage
+                    .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)

walkeeper/src/timeline.rs

@@ -12,6 +12,7 @@ use std::fs::{self, File, OpenOptions};
 use std::io::{Seek, SeekFrom, Write};
 use std::sync::{Arc, Condvar, Mutex};
 use std::time::Duration;
+use zenith_metrics::{register_histogram_vec, Histogram, HistogramVec, DISK_WRITE_SECONDS_BUCKETS};
 use zenith_utils::bin_ser::LeSer;
 use zenith_utils::lsn::Lsn;
 use zenith_utils::zid::{ZTenantId, ZTimelineId};
@@ -73,6 +74,23 @@ pub enum CreateControlFile {
     False,
 }
 
+lazy_static! {
+    static ref PERSIST_SYNC_CONTROL_FILE_SECONDS: HistogramVec = register_histogram_vec!(
+        "safekeeper_persist_sync_control_file_seconds",
+        "Seconds to persist and sync control file, grouped by timeline",
+        &["timeline_id"],
+        DISK_WRITE_SECONDS_BUCKETS.to_vec()
+    )
+    .expect("Failed to register safekeeper_persist_sync_control_file_seconds histogram vec");
+    static ref PERSIST_NOSYNC_CONTROL_FILE_SECONDS: HistogramVec = register_histogram_vec!(
+        "safekeeper_persist_nosync_control_file_seconds",
+        "Seconds to persist and sync control file, grouped by timeline",
+        &["timeline_id"],
+        DISK_WRITE_SECONDS_BUCKETS.to_vec()
+    )
+    .expect("Failed to register safekeeper_persist_nosync_control_file_seconds histogram vec");
+}
+
 impl SharedState {
     /// Get combined stateof all alive replicas
     pub fn get_replicas_state(&self) -> ReplicaState {
@@ -109,9 +127,14 @@ impl SharedState {
         create: CreateControlFile,
     ) -> Result<Self> {
         let (cf, state) = SharedState::load_control_file(conf, timelineid, create)?;
+        let timelineid_str = format!("{}", timelineid);
         let storage = FileStorage {
             control_file: cf,
             conf: conf.clone(),
+            persist_sync_control_file_seconds: PERSIST_SYNC_CONTROL_FILE_SECONDS
+                .with_label_values(&[&timelineid_str]),
+            persist_nosync_control_file_seconds: PERSIST_NOSYNC_CONTROL_FILE_SECONDS
+                .with_label_values(&[&timelineid_str]),
         };
         let (flush_lsn, tli) = if state.server.wal_seg_size != 0 {
             let wal_dir = conf.workdir.join(format!("{}", timelineid));
@@ -376,10 +399,18 @@ impl GlobalTimelines {
 struct FileStorage {
     control_file: File,
     conf: SafeKeeperConf,
+    persist_sync_control_file_seconds: Histogram,
+    persist_nosync_control_file_seconds: Histogram,
 }
 
 impl Storage for FileStorage {
     fn persist(&mut self, s: &SafeKeeperState, sync: bool) -> Result<()> {
+        let _timer = if sync {
+            &self.persist_sync_control_file_seconds
+        } else {
+            &self.persist_nosync_control_file_seconds
+        }
+        .start_timer();
         self.control_file.seek(SeekFrom::Start(0))?;
         s.ser_into(&mut self.control_file)?;
         if sync {

walkeeper/src/wal_service.rs

@@ -18,11 +18,15 @@ pub fn thread_main(conf: SafeKeeperConf, listener: TcpListener) -> Result<()> {
             Ok((socket, peer_addr)) => {
                 debug!("accepted connection from {}", peer_addr);
                 let conf = conf.clone();
-                thread::spawn(move || {
-                    if let Err(err) = handle_socket(socket, conf) {
-                        error!("connection handler exited: {}", err);
-                    }
-                });
+
+                let _ = thread::Builder::new()
+                    .name("WAL service thread".into())
+                    .spawn(move || {
+                        if let Err(err) = handle_socket(socket, conf) {
+                            error!("connection handler exited: {}", err);
+                        }
+                    })
+                    .unwrap();
             }
             Err(e) => error!("Failed to accept connection: {}", e),
         }

zenith/src/main.rs

@@ -203,96 +203,38 @@ fn main() -> Result<()> {
         )
         .get_matches();
 
-    // Create config file
-    if let ("init", Some(init_match)) = matches.subcommand() {
-        let toml_file: String = if let Some(config_path) = init_match.value_of("config") {
-            // load and parse the file
-            std::fs::read_to_string(std::path::Path::new(config_path))
-                .with_context(|| format!("Could not read configuration file \"{}\"", config_path))?
-        } else {
-            // Built-in default config
-            default_conf()
+    let (sub_name, sub_args) = matches.subcommand();
+    let sub_args = sub_args.expect("no subcommand");
+
+    // Check for 'zenith init' command first.
+    let subcmd_result = if sub_name == "init" {
+        handle_init(sub_args)
+    } else {
+        // all other commands need an existing config
+        let env = match LocalEnv::load_config() {
+            Ok(conf) => conf,
+            Err(e) => {
+                eprintln!("Error loading config: {}", e);
+                exit(1);
+            }
         };
 
-        let mut env = LocalEnv::create_config(&toml_file)
-            .with_context(|| "Failed to create zenith configuration")?;
-        env.init()
-            .with_context(|| "Failed to initialize zenith repository")?;
+        match sub_name {
+            "tenant" => handle_tenant(sub_args, &env),
+            "branch" => handle_branch(sub_args, &env),
+            "start" => handle_start_all(sub_args, &env),
+            "stop" => handle_stop_all(sub_args, &env),
+            "pageserver" => handle_pageserver(sub_args, &env),
+            "pg" => handle_pg(sub_args, &env),
+            "safekeeper" => handle_safekeeper(sub_args, &env),
+            _ => bail!("unexpected subcommand {}", sub_name),
+        }
+    };
+    if let Err(e) = subcmd_result {
+        eprintln!("command failed: {}", e);
+        exit(1);
     }
 
-    // all other commands would need config
-    let env = match LocalEnv::load_config() {
-        Ok(conf) => conf,
-        Err(e) => {
-            eprintln!("Error loading config: {}", e);
-            exit(1);
-        }
-    };
-
-    match matches.subcommand() {
-        ("init", Some(_sub_m)) => {
-            // The options were handled above already
-            let pageserver = PageServerNode::from_env(&env);
-            if let Err(e) = pageserver.init(
-                // default_tenantid was generated by the `env.init()` call above
-                Some(&env.default_tenantid.unwrap().to_string()),
-            ) {
-                eprintln!("pageserver init failed: {}", e);
-                exit(1);
-            }
-        }
-        ("tenant", Some(args)) => {
-            if let Err(e) = handle_tenant(args, &env) {
-                eprintln!("tenant command failed: {}", e);
-                exit(1);
-            }
-        }
-
-        ("branch", Some(sub_args)) => {
-            if let Err(e) = handle_branch(sub_args, &env) {
-                eprintln!("branch command failed: {}", e);
-                exit(1);
-            }
-        }
-
-        ("start", Some(sub_match)) => {
-            if let Err(e) = handle_start_all(sub_match, &env) {
-                eprintln!("start command failed: {}", e);
-                exit(1);
-            }
-        }
-
-        ("stop", Some(sub_match)) => {
-            if let Err(e) = handle_stop_all(sub_match, &env) {
-                eprintln!("stop command failed: {}", e);
-                exit(1);
-            }
-        }
-
-        ("pageserver", Some(sub_match)) => {
-            if let Err(e) = handle_pageserver(sub_match, &env) {
-                eprintln!("pageserver command failed: {}", e);
-                exit(1);
-            }
-        }
-
-        ("pg", Some(pg_match)) => {
-            if let Err(e) = handle_pg(pg_match, &env) {
-                eprintln!("pg operation failed: {:?}", e);
-                exit(1);
-            }
-        }
-
-        ("safekeeper", Some(sub_match)) => {
-            if let Err(e) = handle_safekeeper(sub_match, &env) {
-                eprintln!("safekeeper command failed: {}", e);
-                exit(1);
-            }
-        }
-
-        _ => {}
-    };
-
     Ok(())
 }
 
@@ -438,6 +380,35 @@ fn get_tenantid(sub_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<ZTe
     }
 }
 
+fn handle_init(init_match: &ArgMatches) -> Result<()> {
+    // Create config file
+    let toml_file: String = if let Some(config_path) = init_match.value_of("config") {
+        // load and parse the file
+        std::fs::read_to_string(std::path::Path::new(config_path))
+            .with_context(|| format!("Could not read configuration file \"{}\"", config_path))?
+    } else {
+        // Built-in default config
+        default_conf()
+    };
+
+    let mut env = LocalEnv::create_config(&toml_file)
+        .with_context(|| "Failed to create zenith configuration")?;
+    env.init()
+        .with_context(|| "Failed to initialize zenith repository")?;
+
+    // Call 'pageserver init'.
+    let pageserver = PageServerNode::from_env(&env);
+    if let Err(e) = pageserver.init(
+        // default_tenantid was generated by the `env.init()` call above
+        Some(&env.default_tenantid.unwrap().to_string()),
+    ) {
+        eprintln!("pageserver init failed: {}", e);
+        exit(1);
+    }
+
+    Ok(())
+}
+
 fn handle_tenant(tenant_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<()> {
     let pageserver = PageServerNode::from_env(env);
     match tenant_match.subcommand() {
@@ -455,7 +426,10 @@ fn handle_tenant(tenant_match: &ArgMatches, env: &local_env::LocalEnv) -> Result
             pageserver.tenant_create(tenantid)?;
             println!("tenant successfully created on the pageserver");
         }
-        _ => {}
+
+        (sub_name, _) => {
+            bail!("Unexpected tenant subcommand '{}'", sub_name)
+        }
     }
     Ok(())
 }
@@ -484,12 +458,16 @@ fn handle_branch(branch_match: &ArgMatches, env: &local_env::LocalEnv) -> Result
 }
 
 fn handle_pg(pg_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<()> {
+    let (sub_name, sub_args) = pg_match.subcommand();
+    let sub_args = sub_args.expect("no pg subcommand");
+
     let mut cplane = ComputeControlPlane::load(env.clone())?;
 
-    match pg_match.subcommand() {
-        ("list", Some(list_match)) => {
-            let tenantid = get_tenantid(list_match, env)?;
+    // All subcommands take an optional --tenantid option
+    let tenantid = get_tenantid(sub_args, env)?;
 
+    match sub_name {
+        "list" => {
             let branch_infos = get_branch_infos(env, &tenantid).unwrap_or_else(|e| {
                 eprintln!("Failed to load branch info: {}", e);
                 HashMap::new()
@@ -519,23 +497,21 @@ fn handle_pg(pg_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<()> {
                 );
             }
         }
-        ("create", Some(create_match)) => {
-            let tenantid = get_tenantid(create_match, env)?;
-            let node_name = create_match.value_of("node").unwrap_or("main");
-            let timeline_name = create_match.value_of("timeline").unwrap_or(node_name);
+        "create" => {
+            let node_name = sub_args.value_of("node").unwrap_or("main");
+            let timeline_name = sub_args.value_of("timeline").unwrap_or(node_name);
 
-            let port: Option<u16> = match create_match.value_of("port") {
+            let port: Option<u16> = match sub_args.value_of("port") {
                 Some(p) => Some(p.parse()?),
                 None => None,
             };
             cplane.new_node(tenantid, node_name, timeline_name, port)?;
         }
-        ("start", Some(start_match)) => {
-            let tenantid = get_tenantid(start_match, env)?;
-            let node_name = start_match.value_of("node").unwrap_or("main");
-            let timeline_name = start_match.value_of("timeline");
+        "start" => {
+            let node_name = sub_args.value_of("node").unwrap_or("main");
+            let timeline_name = sub_args.value_of("timeline");
 
-            let port: Option<u16> = match start_match.value_of("port") {
+            let port: Option<u16> = match sub_args.value_of("port") {
                 Some(p) => Some(p.parse()?),
                 None => None,
             };
@@ -571,10 +547,9 @@ fn handle_pg(pg_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<()> {
                 node.start(&auth_token)?;
             }
         }
-        ("stop", Some(stop_match)) => {
-            let tenantid = get_tenantid(stop_match, env)?;
-            let node_name = stop_match.value_of("node").unwrap_or("main");
-            let destroy = stop_match.is_present("destroy");
+        "stop" => {
+            let node_name = sub_args.value_of("node").unwrap_or("main");
+            let destroy = sub_args.is_present("destroy");
 
             let node = cplane
                 .nodes
@@ -583,7 +558,9 @@ fn handle_pg(pg_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<()> {
             node.stop(destroy)?;
         }
 
-        _ => {}
+        _ => {
+            bail!("Unexpected pg subcommand '{}'", sub_name)
+        }
     }
 
     Ok(())
@@ -622,7 +599,9 @@ fn handle_pageserver(sub_match: &ArgMatches, env: &local_env::LocalEnv) -> Resul
             }
         }
 
-        _ => {}
+        (sub_name, _) => {
+            bail!("Unexpected pageserver subcommand '{}'", sub_name)
+        }
     }
     Ok(())
 }
@@ -636,26 +615,23 @@ fn get_safekeeper(env: &local_env::LocalEnv, name: &str) -> Result<SafekeeperNod
 }
 
 fn handle_safekeeper(sub_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<()> {
-    match sub_match.subcommand() {
-        ("start", Some(sub_match)) => {
-            let node_name = sub_match
-                .value_of("node")
-                .unwrap_or(DEFAULT_SAFEKEEPER_NAME);
-            let safekeeper = get_safekeeper(env, node_name)?;
+    let (sub_name, sub_args) = sub_match.subcommand();
+    let sub_args = sub_args.expect("no safekeeper subcommand");
 
+    // All the commands take an optional safekeeper name argument
+    let node_name = sub_args.value_of("node").unwrap_or(DEFAULT_SAFEKEEPER_NAME);
+    let safekeeper = get_safekeeper(env, node_name)?;
+
+    match sub_name {
+        "start" => {
             if let Err(e) = safekeeper.start() {
                 eprintln!("safekeeper start failed: {}", e);
                 exit(1);
             }
         }
 
-        ("stop", Some(sub_match)) => {
-            let node_name = sub_match
-                .value_of("node")
-                .unwrap_or(DEFAULT_SAFEKEEPER_NAME);
-            let immediate = sub_match.value_of("stop-mode") == Some("immediate");
-
-            let safekeeper = get_safekeeper(env, node_name)?;
+        "stop" => {
+            let immediate = sub_args.value_of("stop-mode") == Some("immediate");
 
             if let Err(e) = safekeeper.stop(immediate) {
                 eprintln!("safekeeper stop failed: {}", e);
@@ -663,15 +639,10 @@ fn handle_safekeeper(sub_match: &ArgMatches, env: &local_env::LocalEnv) -> Resul
             }
         }
 
-        ("restart", Some(sub_match)) => {
-            let node_name = sub_match
-                .value_of("node")
-                .unwrap_or(DEFAULT_SAFEKEEPER_NAME);
+        "restart" => {
+            let immediate = sub_args.value_of("stop-mode") == Some("immediate");
 
-            let safekeeper = get_safekeeper(env, node_name)?;
-
-            //TODO what shutdown strategy should we use here?
-            if let Err(e) = safekeeper.stop(false) {
+            if let Err(e) = safekeeper.stop(immediate) {
                 eprintln!("safekeeper stop failed: {}", e);
                 exit(1);
             }
@@ -682,7 +653,9 @@ fn handle_safekeeper(sub_match: &ArgMatches, env: &local_env::LocalEnv) -> Resul
             }
         }
 
-        _ => {}
+        _ => {
+            bail!("Unexpected safekeeper subcommand '{}'", sub_name)
+        }
     }
     Ok(())
 }

zenith_metrics/src/lib.rs

@@ -74,6 +74,10 @@ lazy_static! {
     .expect("Failed to register maxrss_kb int gauge");
 }
 
+pub const DISK_WRITE_SECONDS_BUCKETS: &[f64] = &[
+    0.000_050, 0.000_100, 0.000_500, 0.001, 0.003, 0.005, 0.01, 0.05, 0.1, 0.3, 0.5,
+];
+
 // Records I/O stats in a "cross-platform" way.
 // Compiles both on macOS and Linux, but current macOS implementation always returns 0 as values for I/O stats.
 // An alternative is to read procfs (`/proc/[pid]/io`) which does not work under macOS at all, hence abandoned.

zenith_utils/Cargo.toml

@@ -19,6 +19,7 @@ thiserror = "1.0"
 tokio = "1.11"
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+nix = "0.23.0"
 
 zenith_metrics = { path = "../zenith_metrics" }
 workspace_hack = { path = "../workspace_hack" }

zenith_utils/src/batch_fsync.rs

@@ -0,0 +1,43 @@
+use std::{fs::File, io};
+
+const MAX_PENDING_FILES: usize = 100;
+
+#[derive(Default)]
+pub struct BatchFsync {
+    pending: Vec<File>,
+    done: bool,
+}
+
+impl BatchFsync {
+    pub fn add(&mut self, file: File) -> io::Result<()> {
+        if self.pending.len() == MAX_PENDING_FILES {
+            self.sync_batch()?;
+        }
+
+        self.pending.push(file);
+
+        Ok(())
+    }
+
+    /// Must be called before drop.
+    pub fn done(mut self) -> io::Result<()> {
+        self.done = true;
+        self.sync_batch()
+    }
+
+    fn sync_batch(&mut self) -> io::Result<()> {
+        // TODO parallelize
+        for pending_file in self.pending.drain(..) {
+            pending_file.sync_all()?;
+        }
+        self.pending.clear();
+
+        Ok(())
+    }
+}
+
+impl Drop for BatchFsync {
+    fn drop(&mut self) {
+        assert!(self.done);
+    }
+}

zenith_utils/src/lib.rs

@@ -40,3 +40,9 @@ pub mod logging;
 
 // Misc
 pub mod accum;
+
+// Utility for binding TcpListeners with proper socket options.
+pub mod tcp_listener;
+
+// Call fsync for many files at once.
+pub mod batch_fsync;

zenith_utils/src/tcp_listener.rs

@@ -0,0 +1,16 @@
+use std::{
+    io,
+    net::{TcpListener, ToSocketAddrs},
+    os::unix::prelude::AsRawFd,
+};
+
+use nix::sys::socket::{setsockopt, sockopt::ReuseAddr};
+
+/// Bind a [`TcpListener`] to addr with `SO_REUSEADDR` set to true.
+pub fn bind<A: ToSocketAddrs>(addr: A) -> io::Result<TcpListener> {
+    let listener = TcpListener::bind(addr)?;
+
+    setsockopt(listener.as_raw_fd(), ReuseAddr, &true)?;
+
+    Ok(listener)
+}