big batches, doesn't change much

DatadirModification

Cargo.lock

@@ -6850,6 +6850,7 @@ dependencies = [
  "serde_path_to_error",
  "serde_with",
  "signal-hook",
+ "smallvec",
  "strum",
  "strum_macros",
  "thiserror",

Dockerfile

@@ -93,14 +93,13 @@ COPY --from=pg-build /home/nonroot/postgres_install.tar.gz /data/
 
 # By default, pageserver uses `.neon/` working directory in WORKDIR, so create one and fill it with the dummy config.
 # Now, when `docker run ... pageserver` is run, it can start without errors, yet will have some default dummy values.
-RUN mkdir -p /data/.neon/ && \
-  echo "id=1234" > "/data/.neon/identity.toml" && \
-  echo "broker_endpoint='http://storage_broker:50051'\n" \
-       "pg_distrib_dir='/usr/local/'\n" \
-       "listen_pg_addr='0.0.0.0:6400'\n" \
-       "listen_http_addr='0.0.0.0:9898'\n" \
-  > /data/.neon/pageserver.toml && \
-  chown -R neon:neon /data/.neon
+RUN mkdir -p /data/.neon/ && chown -R neon:neon /data/.neon/ \
+    && /usr/local/bin/pageserver -D /data/.neon/ --init \
+       -c "id=1234" \
+       -c "broker_endpoint='http://storage_broker:50051'" \
+       -c "pg_distrib_dir='/usr/local/'" \
+       -c "listen_pg_addr='0.0.0.0:6400'" \
+       -c "listen_http_addr='0.0.0.0:9898'"
 
 # When running a binary that links with libpq, default to using our most recent postgres version.  Binaries
 # that want a particular postgres version will select it explicitly: this is just a default.
@@ -111,6 +110,3 @@ VOLUME ["/data"]
 USER neon
 EXPOSE 6400
 EXPOSE 9898
-
-CMD /usr/local/bin/pageserver -D /data/.neon
-

control_plane/src/pageserver.rs

@@ -25,7 +25,6 @@ use pageserver_client::mgmt_api;
 use postgres_backend::AuthType;
 use postgres_connection::{parse_host_port, PgConnectionConfig};
 use utils::auth::{Claims, Scope};
-use utils::id::NodeId;
 use utils::{
     id::{TenantId, TimelineId},
     lsn::Lsn,
@@ -75,10 +74,6 @@ impl PageServerNode {
         }
     }
 
-    fn pageserver_make_identity_toml(&self, node_id: NodeId) -> toml_edit::Document {
-        toml_edit::Document::from_str(&format!("id={node_id}")).unwrap()
-    }
-
     fn pageserver_init_make_toml(
         &self,
         conf: NeonLocalInitPageserverConf,
@@ -191,19 +186,6 @@ impl PageServerNode {
             .write_all(config.to_string().as_bytes())
             .context("write pageserver toml")?;
         drop(config_file);
-
-        let identity_file_path = datadir.join("identity.toml");
-        let mut identity_file = std::fs::OpenOptions::new()
-            .create_new(true)
-            .write(true)
-            .open(identity_file_path)
-            .with_context(|| format!("open identity toml for write: {config_file_path:?}"))?;
-        let identity_toml = self.pageserver_make_identity_toml(node_id);
-        identity_file
-            .write_all(identity_toml.to_string().as_bytes())
-            .context("write identity toml")?;
-        drop(identity_toml);
-
         // TODO: invoke a TBD config-check command to validate that pageserver will start with the written config
 
         // Write metadata file, used by pageserver on startup to register itself with

docker-compose/compute_wrapper/shell/compute.sh

@@ -33,7 +33,7 @@ echo $result | jq .
 
 generate_id timeline_id
 PARAMS=(
-     -sbf
+     -sb 
      -X POST
      -H "Content-Type: application/json"
      -d "{\"new_timeline_id\": \"${timeline_id}\", \"pg_version\": ${PG_VERSION}}"

docker-compose/docker-compose.yml

@@ -31,14 +31,25 @@ services:
     restart: always
     image: ${REPOSITORY:-neondatabase}/neon:${TAG:-latest}
     environment:
+      - BROKER_ENDPOINT='http://storage_broker:50051'
       - AWS_ACCESS_KEY_ID=minio
       - AWS_SECRET_ACCESS_KEY=password
       #- RUST_BACKTRACE=1
     ports:
        #- 6400:6400  # pg protocol handler
        - 9898:9898 # http endpoints
-    volumes:
-      - ./pageserver_config:/data/.neon/
+    entrypoint:
+      - "/bin/sh"
+      - "-c"
+    command:
+      - "/usr/local/bin/pageserver -D /data/.neon/
+                                   -c \"broker_endpoint=$$BROKER_ENDPOINT\"
+                                   -c \"listen_pg_addr='0.0.0.0:6400'\"
+                                   -c \"listen_http_addr='0.0.0.0:9898'\"
+                                   -c \"remote_storage={endpoint='http://minio:9000',
+                                                        bucket_name='neon',
+                                                        bucket_region='eu-north-1',
+                                                        prefix_in_bucket='/pageserver/'}\""
     depends_on:
       - storage_broker
       - minio_create_buckets

docker-compose/pageserver_config/identity.toml

@@ -1 +0,0 @@
-id=1234

docker-compose/pageserver_config/pageserver.toml

@@ -1,5 +0,0 @@
-broker_endpoint='http://storage_broker:50051'
-pg_distrib_dir='/usr/local/'
-listen_pg_addr='0.0.0.0:6400'
-listen_http_addr='0.0.0.0:9898'
-remote_storage={ endpoint='http://minio:9000', bucket_name='neon', bucket_region='eu-north-1', prefix_in_bucket='/pageserver' }

libs/pageserver_api/src/key.rs

@@ -12,7 +12,7 @@ use crate::reltag::{BlockNumber, RelTag, SlruKind};
 ///
 /// The Repository treats this as an opaque struct, but see the code in pgdatadir_mapping.rs
 /// for what we actually store in these fields.
-#[derive(Debug, Clone, Copy, Hash, PartialEq, Eq, Ord, PartialOrd, Serialize, Deserialize)]
+#[derive(Debug, Clone, Copy, Hash, Eq, PartialEq, Ord, Serialize, Deserialize)]
 pub struct Key {
     pub field1: u8,
     pub field2: u32,
@@ -22,6 +22,41 @@ pub struct Key {
     pub field6: u32,
 }
 
+impl PartialOrd for Key {
+    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
+        if self.field1 == other.field1
+            && self.field2 == other.field2
+            && self.field3 == other.field3
+            && self.field4 == other.field4
+            && self.field5 == other.field5
+        {
+            self.field6.partial_cmp(&other.field6)
+        } else {
+            match self.field1.partial_cmp(&other.field1) {
+                Some(core::cmp::Ordering::Equal) => {}
+                ord => return ord,
+            }
+            match self.field2.partial_cmp(&other.field2) {
+                Some(core::cmp::Ordering::Equal) => {}
+                ord => return ord,
+            }
+            match self.field3.partial_cmp(&other.field3) {
+                Some(core::cmp::Ordering::Equal) => {}
+                ord => return ord,
+            }
+            match self.field4.partial_cmp(&other.field4) {
+                Some(core::cmp::Ordering::Equal) => {}
+                ord => return ord,
+            }
+            match self.field5.partial_cmp(&other.field5) {
+                Some(core::cmp::Ordering::Equal) => {}
+                ord => return ord,
+            }
+            self.field6.partial_cmp(&other.field6)
+        }
+    }
+}
+
 /// The storage key size.
 pub const KEY_SIZE: usize = 18;
 

libs/postgres_ffi/src/lib.rs

@@ -132,7 +132,7 @@ pub const RELSEG_SIZE: u32 = 1024 * 1024 * 1024 / (BLCKSZ as u32);
 pub const XLOG_BLCKSZ: usize = 8192;
 pub const WAL_SEGMENT_SIZE: usize = 16 * 1024 * 1024;
 
-pub const MAX_SEND_SIZE: usize = XLOG_BLCKSZ * 16;
+pub const MAX_SEND_SIZE: usize = XLOG_BLCKSZ * 128;
 
 // Export some version independent functions that are used outside of this mod
 pub use v14::xlog_utils::encode_logical_message;

libs/utils/Cargo.toml

@@ -36,6 +36,7 @@ routerify.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 signal-hook.workspace = true
+smallvec.workspace = true
 thiserror.workspace = true
 tokio.workspace = true
 tokio-tar.workspace = true

libs/utils/src/vec_map.rs

@@ -1,11 +1,15 @@
 use std::{alloc::Layout, cmp::Ordering, ops::RangeBounds};
 
+use smallvec::SmallVec;
+
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub enum VecMapOrdering {
     Greater,
     GreaterOrEqual,
 }
 
+const INLINE_ELEMENTS: usize = 1;
+
 /// Ordered map datastructure implemented in a Vec.
 /// Append only - can only add keys that are larger than the
 /// current max key.
@@ -13,7 +17,7 @@ pub enum VecMapOrdering {
 /// during `VecMap` construction.
 #[derive(Clone, Debug)]
 pub struct VecMap<K, V> {
-    data: Vec<(K, V)>,
+    data: SmallVec<[(K, V); INLINE_ELEMENTS]>,
     ordering: VecMapOrdering,
 }
 
@@ -37,14 +41,18 @@ pub enum VecMapError {
 impl<K: Ord, V> VecMap<K, V> {
     pub fn new(ordering: VecMapOrdering) -> Self {
         Self {
-            data: Vec::new(),
+            data: Default::default(),
             ordering,
         }
     }
 
+    pub fn len(&self) -> usize {
+        self.data.len()
+    }
+
     pub fn with_capacity(capacity: usize, ordering: VecMapOrdering) -> Self {
         Self {
-            data: Vec::with_capacity(capacity),
+            data: SmallVec::with_capacity(capacity),
             ordering,
         }
     }
@@ -119,6 +127,11 @@ impl<K: Ord, V> VecMap<K, V> {
         Ok((None, delta_size))
     }
 
+    /// Where the key is known to be unique, and we don't want any instrumentation
+    pub fn append2(&mut self, key: K, value: V) {
+        self.data.push((key, value));
+    }
+
     /// Split the map into two.
     ///
     /// The left map contains everything before `cutoff` (exclusive).
@@ -135,11 +148,11 @@ impl<K: Ord, V> VecMap<K, V> {
 
         (
             VecMap {
-                data: self.data[..split_idx].to_vec(),
+                data: SmallVec::from(&self.data[..split_idx]),
                 ordering: self.ordering,
             },
             VecMap {
-                data: self.data[split_idx..].to_vec(),
+                data: SmallVec::from(&self.data[split_idx..]),
                 ordering: self.ordering,
             },
         )
@@ -186,7 +199,10 @@ impl<K: Ord, V> VecMap<K, V> {
     /// Instrument an operation on the underlying [`Vec`].
     /// Will panic if the operation decreases capacity.
     /// Returns the increase in memory usage caused by the op.
-    fn instrument_vec_op(&mut self, op: impl FnOnce(&mut Vec<(K, V)>)) -> usize {
+    fn instrument_vec_op(
+        &mut self,
+        op: impl FnOnce(&mut SmallVec<[(K, V); INLINE_ELEMENTS]>),
+    ) -> usize {
         let old_cap = self.data.capacity();
         op(&mut self.data);
         let new_cap = self.data.capacity();
@@ -226,7 +242,7 @@ impl<K: Ord, V> VecMap<K, V> {
 
 impl<K: Ord, V> IntoIterator for VecMap<K, V> {
     type Item = (K, V);
-    type IntoIter = std::vec::IntoIter<(K, V)>;
+    type IntoIter = smallvec::IntoIter<[(K, V); INLINE_ELEMENTS]>;
 
     fn into_iter(self) -> Self::IntoIter {
         self.data.into_iter()

pageserver/src/bin/pageserver.rs

@@ -2,18 +2,17 @@
 
 //! Main entry point for the Page Server executable.
 
-use std::env;
 use std::env::{var, VarError};
 use std::io::Read;
 use std::sync::Arc;
 use std::time::Duration;
+use std::{env, ops::ControlFlow, str::FromStr};
 
 use anyhow::{anyhow, Context};
 use camino::Utf8Path;
 use clap::{Arg, ArgAction, Command};
 
 use metrics::launch_timestamp::{set_launch_timestamp_metric, LaunchTimestamp};
-use pageserver::config::PageserverIdentity;
 use pageserver::control_plane_client::ControlPlaneClient;
 use pageserver::disk_usage_eviction_task::{self, launch_disk_usage_global_eviction_task};
 use pageserver::metrics::{STARTUP_DURATION, STARTUP_IS_LOADING};
@@ -26,7 +25,7 @@ use tracing::*;
 
 use metrics::set_build_info_metric;
 use pageserver::{
-    config::PageServerConf,
+    config::{defaults::*, PageServerConf},
     context::{DownloadBehavior, RequestContext},
     deletion_queue::DeletionQueue,
     http, page_cache, page_service, task_mgr,
@@ -85,13 +84,18 @@ fn main() -> anyhow::Result<()> {
         .with_context(|| format!("Error opening workdir '{workdir}'"))?;
 
     let cfg_file_path = workdir.join("pageserver.toml");
-    let identity_file_path = workdir.join("identity.toml");
 
     // Set CWD to workdir for non-daemon modes
     env::set_current_dir(&workdir)
         .with_context(|| format!("Failed to set application's current dir to '{workdir}'"))?;
 
-    let conf = initialize_config(&identity_file_path, &cfg_file_path, &workdir)?;
+    let conf = match initialize_config(&cfg_file_path, arg_matches, &workdir)? {
+        ControlFlow::Continue(conf) => conf,
+        ControlFlow::Break(()) => {
+            info!("Pageserver config init successful");
+            return Ok(());
+        }
+    };
 
     // Initialize logging.
     //
@@ -146,55 +150,70 @@ fn main() -> anyhow::Result<()> {
 }
 
 fn initialize_config(
-    identity_file_path: &Utf8Path,
     cfg_file_path: &Utf8Path,
+    arg_matches: clap::ArgMatches,
     workdir: &Utf8Path,
-) -> anyhow::Result<&'static PageServerConf> {
-    // The deployment orchestrator writes out an indentity file containing the node id
-    // for all pageservers. This file is the source of truth for the node id. In order
-    // to allow for rolling back pageserver releases, the node id is also included in
-    // the pageserver config that the deployment orchestrator writes to disk for the pageserver.
-    // A rolled back version of the pageserver will get the node id from the pageserver.toml
-    // config file.
-    let identity = match std::fs::File::open(identity_file_path) {
+) -> anyhow::Result<ControlFlow<(), &'static PageServerConf>> {
+    let init = arg_matches.get_flag("init");
+
+    let file_contents: Option<toml_edit::Document> = match std::fs::File::open(cfg_file_path) {
         Ok(mut f) => {
-            let md = f.metadata().context("stat config file")?;
-            if !md.is_file() {
-                anyhow::bail!("Pageserver found identity file but it is a dir entry: {identity_file_path}. Aborting start up ...");
+            if init {
+                anyhow::bail!("config file already exists: {cfg_file_path}");
             }
-
-            let mut s = String::new();
-            f.read_to_string(&mut s).context("read identity file")?;
-            toml_edit::de::from_str::<PageserverIdentity>(&s)?
-        }
-        Err(e) => {
-            anyhow::bail!("Pageserver could not read identity file: {identity_file_path}: {e}. Aborting start up ...");
-        }
-    };
-
-    let config: toml_edit::Document = match std::fs::File::open(cfg_file_path) {
-        Ok(mut f) => {
             let md = f.metadata().context("stat config file")?;
             if md.is_file() {
                 let mut s = String::new();
                 f.read_to_string(&mut s).context("read config file")?;
-                s.parse().context("parse config file toml")?
+                Some(s.parse().context("parse config file toml")?)
             } else {
                 anyhow::bail!("directory entry exists but is not a file: {cfg_file_path}");
             }
         }
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => None,
         Err(e) => {
             anyhow::bail!("open pageserver config: {e}: {cfg_file_path}");
         }
     };
 
-    debug!("Using pageserver toml: {config}");
+    let mut effective_config = file_contents.unwrap_or_else(|| {
+        DEFAULT_CONFIG_FILE
+            .parse()
+            .expect("unit tests ensure this works")
+    });
+
+    // Patch with overrides from the command line
+    if let Some(values) = arg_matches.get_many::<String>("config-override") {
+        for option_line in values {
+            let doc = toml_edit::Document::from_str(option_line).with_context(|| {
+                format!("Option '{option_line}' could not be parsed as a toml document")
+            })?;
+
+            for (key, item) in doc.iter() {
+                effective_config.insert(key, item.clone());
+            }
+        }
+    }
+
+    debug!("Resulting toml: {effective_config}");
 
     // Construct the runtime representation
-    let conf = PageServerConf::parse_and_validate(identity.id, &config, workdir)
+    let conf = PageServerConf::parse_and_validate(&effective_config, workdir)
         .context("Failed to parse pageserver configuration")?;
 
-    Ok(Box::leak(Box::new(conf)))
+    if init {
+        info!("Writing pageserver config to '{cfg_file_path}'");
+
+        std::fs::write(cfg_file_path, effective_config.to_string())
+            .with_context(|| format!("Failed to write pageserver config to '{cfg_file_path}'"))?;
+        info!("Config successfully written to '{cfg_file_path}'")
+    }
+
+    Ok(if init {
+        ControlFlow::Break(())
+    } else {
+        ControlFlow::Continue(Box::leak(Box::new(conf)))
+    })
 }
 
 struct WaitForPhaseResult<F: std::future::Future + Unpin> {
@@ -715,12 +734,28 @@ fn cli() -> Command {
     Command::new("Neon page server")
         .about("Materializes WAL stream to pages and serves them to the postgres")
         .version(version())
+        .arg(
+            Arg::new("init")
+                .long("init")
+                .action(ArgAction::SetTrue)
+                .help("Initialize pageserver with all given config overrides"),
+        )
         .arg(
             Arg::new("workdir")
                 .short('D')
                 .long("workdir")
                 .help("Working directory for the pageserver"),
         )
+        // See `settings.md` for more details on the extra configuration patameters pageserver can process
+        .arg(
+            Arg::new("config-override")
+                .long("config-override")
+                .short('c')
+                .num_args(1)
+                .action(ArgAction::Append)
+                .help("Additional configuration overrides of the ones from the toml config file (or new ones to add there). \
+                Any option has to be a valid toml document, example: `-c=\"foo='hey'\"` `-c=\"foo={value=1}\"`"),
+        )
         .arg(
             Arg::new("enabled-features")
                 .long("enabled-features")

pageserver/src/config.rs

@@ -7,8 +7,8 @@
 use anyhow::{anyhow, bail, ensure, Context, Result};
 use pageserver_api::{models::ImageCompressionAlgorithm, shard::TenantShardId};
 use remote_storage::{RemotePath, RemoteStorageConfig};
+use serde;
 use serde::de::IntoDeserializer;
-use serde::{self, Deserialize};
 use std::env;
 use storage_broker::Uri;
 use utils::crashsafe::path_with_suffix_extension;
@@ -406,13 +406,6 @@ struct PageServerConfigBuilder {
 }
 
 impl PageServerConfigBuilder {
-    fn new(node_id: NodeId) -> Self {
-        let mut this = Self::default();
-        this.id(node_id);
-
-        this
-    }
-
     #[inline(always)]
     fn default_values() -> Self {
         use self::BuilderValue::*;
@@ -888,12 +881,8 @@ impl PageServerConf {
     /// validating the input and failing on errors.
     ///
     /// This leaves any options not present in the file in the built-in defaults.
-    pub fn parse_and_validate(
-        node_id: NodeId,
-        toml: &Document,
-        workdir: &Utf8Path,
-    ) -> anyhow::Result<Self> {
-        let mut builder = PageServerConfigBuilder::new(node_id);
+    pub fn parse_and_validate(toml: &Document, workdir: &Utf8Path) -> anyhow::Result<Self> {
+        let mut builder = PageServerConfigBuilder::default();
         builder.workdir(workdir.to_owned());
 
         let mut t_conf = TenantConfOpt::default();
@@ -924,8 +913,7 @@ impl PageServerConf {
                 "tenant_config" => {
                     t_conf = TenantConfOpt::try_from(item.to_owned()).context(format!("failed to parse: '{key}'"))?;
                 }
-                "id" => {}, // Ignoring `id` field in pageserver.toml - using identity.toml as the source of truth
-                            // Logging is not set up yet, so we can't do it.
+                "id" => builder.id(NodeId(parse_toml_u64(key, item)?)),
                 "broker_endpoint" => builder.broker_endpoint(parse_toml_string(key, item)?.parse().context("failed to parse broker endpoint")?),
                 "broker_keepalive_interval" => builder.broker_keepalive_interval(parse_toml_duration(key, item)?),
                 "log_format" => builder.log_format(
@@ -1102,12 +1090,6 @@ impl PageServerConf {
     }
 }
 
-#[derive(Deserialize)]
-#[serde(deny_unknown_fields)]
-pub struct PageserverIdentity {
-    pub id: NodeId,
-}
-
 // Helper functions to parse a toml Item
 
 fn parse_toml_string(name: &str, item: &Item) -> Result<String> {
@@ -1277,7 +1259,7 @@ background_task_maximum_delay = '334 s'
         );
         let toml = config_string.parse()?;
 
-        let parsed_config = PageServerConf::parse_and_validate(NodeId(10), &toml, &workdir)
+        let parsed_config = PageServerConf::parse_and_validate(&toml, &workdir)
             .unwrap_or_else(|e| panic!("Failed to parse config '{config_string}', reason: {e:?}"));
 
         assert_eq!(
@@ -1359,7 +1341,7 @@ background_task_maximum_delay = '334 s'
         );
         let toml = config_string.parse()?;
 
-        let parsed_config = PageServerConf::parse_and_validate(NodeId(10), &toml, &workdir)
+        let parsed_config = PageServerConf::parse_and_validate(&toml, &workdir)
             .unwrap_or_else(|e| panic!("Failed to parse config '{config_string}', reason: {e:?}"));
 
         assert_eq!(
@@ -1449,13 +1431,12 @@ broker_endpoint = '{broker_endpoint}'
 
             let toml = config_string.parse()?;
 
-            let parsed_remote_storage_config =
-                PageServerConf::parse_and_validate(NodeId(10), &toml, &workdir)
-                    .unwrap_or_else(|e| {
-                        panic!("Failed to parse config '{config_string}', reason: {e:?}")
-                    })
-                    .remote_storage_config
-                    .expect("Should have remote storage config for the local FS");
+            let parsed_remote_storage_config = PageServerConf::parse_and_validate(&toml, &workdir)
+                .unwrap_or_else(|e| {
+                    panic!("Failed to parse config '{config_string}', reason: {e:?}")
+                })
+                .remote_storage_config
+                .expect("Should have remote storage config for the local FS");
 
             assert_eq!(
                 parsed_remote_storage_config,
@@ -1511,13 +1492,12 @@ broker_endpoint = '{broker_endpoint}'
 
             let toml = config_string.parse()?;
 
-            let parsed_remote_storage_config =
-                PageServerConf::parse_and_validate(NodeId(10), &toml, &workdir)
-                    .unwrap_or_else(|e| {
-                        panic!("Failed to parse config '{config_string}', reason: {e:?}")
-                    })
-                    .remote_storage_config
-                    .expect("Should have remote storage config for S3");
+            let parsed_remote_storage_config = PageServerConf::parse_and_validate(&toml, &workdir)
+                .unwrap_or_else(|e| {
+                    panic!("Failed to parse config '{config_string}', reason: {e:?}")
+                })
+                .remote_storage_config
+                .expect("Should have remote storage config for S3");
 
             assert_eq!(
                 parsed_remote_storage_config,
@@ -1596,7 +1576,7 @@ threshold = "20m"
 "#,
         );
         let toml: Document = pageserver_conf_toml.parse()?;
-        let conf = PageServerConf::parse_and_validate(NodeId(333), &toml, &workdir)?;
+        let conf = PageServerConf::parse_and_validate(&toml, &workdir)?;
 
         assert_eq!(conf.pg_distrib_dir, pg_distrib_dir);
         assert_eq!(
@@ -1612,11 +1592,7 @@ threshold = "20m"
                 .evictions_low_residence_duration_metric_threshold,
             Duration::from_secs(20 * 60)
         );
-
-        // Assert that the node id provided by the indentity file (threaded
-        // through the call to [`PageServerConf::parse_and_validate`] is
-        // used.
-        assert_eq!(conf.id, NodeId(333));
+        assert_eq!(conf.id, NodeId(222));
         assert_eq!(
             conf.disk_usage_based_eviction,
             Some(DiskUsageEvictionTaskConfig {
@@ -1661,7 +1637,7 @@ threshold = "20m"
 "#,
         );
         let toml: Document = pageserver_conf_toml.parse().unwrap();
-        let conf = PageServerConf::parse_and_validate(NodeId(222), &toml, &workdir).unwrap();
+        let conf = PageServerConf::parse_and_validate(&toml, &workdir).unwrap();
 
         match &conf.default_tenant_conf.eviction_policy {
             EvictionPolicy::OnlyImitiate(t) => {
@@ -1680,7 +1656,7 @@ threshold = "20m"
 remote_storage = {}
         "#;
         let doc = toml_edit::Document::from_str(input).unwrap();
-        let err = PageServerConf::parse_and_validate(NodeId(222), &doc, &workdir)
+        let err = PageServerConf::parse_and_validate(&doc, &workdir)
             .expect_err("empty remote_storage field should fail, don't specify it if you want no remote_storage");
         assert!(format!("{err}").contains("remote_storage"), "{err}");
     }

pageserver/src/pgdatadir_mapping.rs

@@ -174,6 +174,7 @@ impl Timeline {
             pending_deletions: Vec::new(),
             pending_nblocks: 0,
             pending_directory_entries: Vec::new(),
+            latest_rel_sizes: Default::default(),
             lsn,
         }
     }
@@ -1045,6 +1046,11 @@ pub struct DatadirModification<'a> {
     pending_deletions: Vec<(Range<Key>, Lsn)>,
     pending_nblocks: i64,
 
+    // We update relation sizes when appending.  Since writing is single threaded, once we
+    // have updated a relation size we may be sure that its size is unchanged within the
+    // same DatadirModification
+    latest_rel_sizes: HashMap<RelTag, u32>,
+
     /// For special "directory" keys that store key-value maps, track the size of the map
     /// if it was updated in this modification.
     pending_directory_entries: Vec<(DirectoryKind, usize)>,
@@ -1407,7 +1413,10 @@ impl<'a> DatadirModification<'a> {
 
         // Put size
         let size_key = rel_size_to_key(rel);
-        let old_size = self.get(size_key, ctx).await?.get_u32_le();
+        let old_size = match self.latest_rel_sizes.get(&rel) {
+            Some(s) => *s,
+            None => self.get(size_key, ctx).await?.get_u32_le(),
+        };
 
         // only extend relation here. never decrease the size
         if nblocks > old_size {
@@ -1418,6 +1427,8 @@ impl<'a> DatadirModification<'a> {
             self.tline.set_cached_rel_size(rel, self.lsn, nblocks);
 
             self.pending_nblocks += nblocks as i64 - old_size as i64;
+
+            self.latest_rel_sizes.insert(rel, nblocks);
         }
         Ok(())
     }

pageserver/src/tenant/ephemeral_file/zero_padded_read_write.rs

@@ -28,7 +28,7 @@ use crate::{
     },
 };
 
-const TAIL_SZ: usize = 64 * 1024;
+const TAIL_SZ: usize = 4096 * 1024;
 
 /// See module-level comment.
 pub struct RW<W: OwnedAsyncWriter> {

pageserver/src/tenant/storage_layer/inmemory_layer.rs

@@ -521,6 +521,30 @@ impl InMemoryLayer {
         self.put_value_locked(&mut inner, key, lsn, buf, ctx).await
     }
 
+    pub(crate) async fn put_values(
+        &self,
+        mut values: Vec<(Lsn, Key, smallvec::SmallVec<[u8; 256]>, u64)>,
+        ctx: &RequestContext,
+    ) -> Result<()> {
+        let mut inner = self.inner.write().await;
+        self.assert_writable();
+        for (_lsn, _key, buf, off) in &mut values {
+            *off = self.put_value_locked2(&mut inner, &buf, ctx).await?;
+        }
+
+        for (lsn, key, _buf, off) in values.into_iter() {
+            let vec_map = inner.index.entry(key).or_default();
+
+            // Use fast version of append, since we know our LSNs are already sorted
+            vec_map.append2(lsn, off);
+        }
+
+        let size = inner.file.len();
+        inner.resource_units.maybe_publish_size(size);
+
+        Ok(())
+    }
+
     async fn put_value_locked(
         &self,
         locked_inner: &mut RwLockWriteGuard<'_, InMemoryLayerInner>,
@@ -556,6 +580,27 @@ impl InMemoryLayer {
         Ok(())
     }
 
+    async fn put_value_locked2(
+        &self,
+        locked_inner: &mut RwLockWriteGuard<'_, InMemoryLayerInner>,
+        buf: &[u8],
+        ctx: &RequestContext,
+    ) -> Result<u64> {
+        let off = {
+            locked_inner
+                .file
+                .write_blob(
+                    buf,
+                    &RequestContextBuilder::extend(ctx)
+                        .page_content_kind(PageContentKind::InMemoryLayer)
+                        .build(),
+                )
+                .await?
+        };
+
+        Ok(off)
+    }
+
     pub(crate) fn get_opened_at(&self) -> Instant {
         self.opened_at
     }
@@ -574,8 +619,6 @@ impl InMemoryLayer {
     /// Records the end_lsn for non-dropped layers.
     /// `end_lsn` is exclusive
     pub async fn freeze(&self, end_lsn: Lsn) {
-        let inner = self.inner.write().await;
-
         assert!(
             self.start_lsn < end_lsn,
             "{} >= {}",
@@ -593,9 +636,13 @@ impl InMemoryLayer {
             })
             .expect("frozen_local_path_str set only once");
 
-        for vec_map in inner.index.values() {
-            for (lsn, _pos) in vec_map.as_slice() {
-                assert!(*lsn < end_lsn);
+        #[cfg(debug_assertions)]
+        {
+            let inner = self.inner.write().await;
+            for vec_map in inner.index.values() {
+                for (lsn, _pos) in vec_map.as_slice() {
+                    debug_assert!(*lsn < end_lsn);
+                }
             }
         }
     }

pageserver/src/tenant/timeline.rs

@@ -5990,10 +5990,45 @@ impl<'a> TimelineWriter<'a> {
         batch: VecMap<Lsn, (Key, Value)>,
         ctx: &RequestContext,
     ) -> anyhow::Result<()> {
-        for (lsn, (key, val)) in batch {
-            self.put(key, lsn, &val, ctx).await?
+        if batch.is_empty() {
+            return Ok(());
         }
 
+        let first_lsn = batch.as_slice().first().unwrap().0;
+        let last_lsn = batch.as_slice().last().unwrap().0;
+        let mut total_serialized_size = 0;
+
+        let mut serialized = Vec::with_capacity(batch.len());
+        for (l, (k, v)) in batch.into_iter() {
+            // Avoid doing allocations for "small" values.
+            // In the regression test suite, the limit of 256 avoided allocations in 95% of cases:
+            // https://github.com/neondatabase/neon/pull/5056#discussion_r1301975061
+            let mut buf = smallvec::SmallVec::<[u8; 256]>::new();
+            v.ser_into(&mut buf)
+                .expect("Serialization of Value is infallible");
+            let buf_size: u64 = buf.len().try_into().expect("oversized value buf");
+            total_serialized_size += buf_size;
+            serialized.push((l, k, buf, 0));
+        }
+
+        let action = self.get_open_layer_action(first_lsn, total_serialized_size);
+        let layer = self
+            .handle_open_layer_action(first_lsn, action, ctx)
+            .await?;
+
+        layer.put_values(serialized, ctx).await?;
+
+        // Update the current size only when the entire write was ok.
+        // In case of failures, we may have had partial writes which
+        // render the size tracking out of sync. That's ok because
+        // the checkpoint distance should be significantly smaller
+        // than the S3 single shot upload limit of 5GiB.
+        let state = self.write_guard.as_mut().unwrap();
+
+        state.current_size += total_serialized_size;
+        state.prev_lsn = Some(last_lsn);
+        state.max_lsn = std::cmp::max(state.max_lsn, Some(last_lsn));
+
         Ok(())
     }
 

safekeeper/src/send_wal.rs

@@ -456,7 +456,7 @@ impl SafekeeperPostgresHandler {
         // not synchronized with sends, so this avoids deadlocks.
         let reader = pgb.split().context("START_REPLICATION split")?;
 
-        let mut sender = WalSender {
+        let mut sender = Box::new(WalSender {
             pgb,
             // should succeed since we're already holding another guard
             tli: tli.wal_residence_guard().await?,
@@ -468,7 +468,7 @@ impl SafekeeperPostgresHandler {
             ws_guard: ws_guard.clone(),
             wal_reader,
             send_buf: [0; MAX_SEND_SIZE],
-        };
+        });
         let mut reply_reader = ReplyReader {
             reader,
             ws_guard: ws_guard.clone(),
@@ -586,6 +586,7 @@ impl<IO: AsyncRead + AsyncWrite + Unpin> WalSender<'_, IO> {
             }
             let send_size = (chunk_end_pos.0 - self.start_pos.0) as usize;
             let send_buf = &mut self.send_buf[..send_size];
+
             let send_size: usize;
             {
                 // If uncommitted part is being pulled, check that the term is

test_runner/fixtures/compare_fixtures.py

@@ -45,6 +45,12 @@ class PgCompare(ABC):
     def flush(self):
         pass
 
+    def flush1(self):
+        _x = 1 + 2
+
+    def flush2(self):
+        _x = 1 + 2
+
     @abstractmethod
     def report_peak_memory_use(self):
         pass
@@ -130,7 +136,13 @@ class NeonCompare(PgCompare):
         return self._pg_bin
 
     def flush(self):
+        self.flush1()
+        self.flush2()
+
+    def flush1(self):
         wait_for_last_flush_lsn(self.env, self._pg, self.tenant, self.timeline)
+
+    def flush2(self):
         self.pageserver_http_client.timeline_checkpoint(self.tenant, self.timeline)
         self.pageserver_http_client.timeline_gc(self.tenant, self.timeline, 0)
 

test_runner/performance/test_bulk_insert.py

@@ -1,9 +1,10 @@
+import time
 from contextlib import closing
 
-import pytest
 from fixtures.benchmark_fixture import MetricReport
 from fixtures.common_types import Lsn
 from fixtures.compare_fixtures import NeonCompare, PgCompare
+from fixtures.log_helper import log
 from fixtures.pg_version import PgVersion
 
 
@@ -17,7 +18,6 @@ from fixtures.pg_version import PgVersion
 # 3. Disk space used
 # 4. Peak memory usage
 #
-@pytest.mark.skip("See https://github.com/neondatabase/neon/issues/7124")
 def test_bulk_insert(neon_with_baseline: PgCompare):
     env = neon_with_baseline
 
@@ -28,10 +28,15 @@ def test_bulk_insert(neon_with_baseline: PgCompare):
             cur.execute("create table huge (i int, j int);")
 
             # Run INSERT, recording the time and I/O it takes
+            log.info("Writing...")
             with env.record_pageserver_writes("pageserver_writes"):
                 with env.record_duration("insert"):
-                    cur.execute("insert into huge values (generate_series(1, 5000000), 0);")
-                    env.flush()
+                    cur.execute("insert into huge values (generate_series(1, 20000000), 0);")
+                    env.flush1()
+
+            log.info("Finished writing")
+
+            env.flush2()
 
             env.report_peak_memory_use()
             env.report_size()
@@ -70,8 +75,12 @@ def measure_recovery_time(env: NeonCompare):
     env.env.pageserver.tenant_create(tenant_id=env.tenant, generation=attach_gen)
 
     # Measure recovery time
-    with env.record_duration("wal_recovery"):
-        client.timeline_create(pg_version, env.tenant, env.timeline)
+    client.timeline_create(pg_version, env.tenant, env.timeline)
 
+    log.info("Recovering...")
+    with env.record_duration("wal_recovery"):
         # Flush, which will also wait for lsn to catch up
-        env.flush()
+        env.flush1()
+
+    log.info("Finished recovering")
+    time.sleep(5)

test_runner/regress/test_pageserver_api.py

@@ -1,5 +1,8 @@
+import subprocess
+from pathlib import Path
 from typing import Optional
 
+import toml
 from fixtures.common_types import Lsn, TenantId, TimelineId
 from fixtures.neon_fixtures import (
     DEFAULT_BRANCH_NAME,
@@ -10,6 +13,67 @@ from fixtures.pageserver.http import PageserverHttpClient
 from fixtures.utils import wait_until
 
 
+def test_pageserver_init_node_id(neon_simple_env: NeonEnv, neon_binpath: Path):
+    """
+    NB: The neon_local doesn't use `--init` mode anymore, but our production
+    deployment still does => https://github.com/neondatabase/aws/pull/1322
+    """
+    workdir = neon_simple_env.pageserver.workdir
+    pageserver_config = workdir / "pageserver.toml"
+    pageserver_bin = neon_binpath / "pageserver"
+
+    def run_pageserver(args):
+        return subprocess.run(
+            [str(pageserver_bin), "-D", str(workdir), *args],
+            check=False,
+            universal_newlines=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+        )
+
+    neon_simple_env.pageserver.stop()
+
+    with open(neon_simple_env.pageserver.config_toml_path, "r") as f:
+        ps_config = toml.load(f)
+
+    required_config_keys = [
+        "pg_distrib_dir",
+        "listen_pg_addr",
+        "listen_http_addr",
+        "pg_auth_type",
+        "http_auth_type",
+        # TODO: only needed for NEON_PAGESERVER_PANIC_ON_UNSPECIFIED_COMPACTION_ALGORITHM in https://github.com/neondatabase/neon/pull/7748
+        # "tenant_config",
+    ]
+    required_config_overrides = [
+        f"--config-override={toml.dumps({k: ps_config[k]})}" for k in required_config_keys
+    ]
+
+    pageserver_config.unlink()
+
+    bad_init = run_pageserver(["--init", *required_config_overrides])
+    assert (
+        bad_init.returncode == 1
+    ), "pageserver should not be able to init new config without the node id"
+    assert 'missing config value "id"' in bad_init.stderr
+    assert not pageserver_config.exists(), "config file should not be created after init error"
+
+    good_init_cmd = [
+        "--init",
+        f"--config-override=id={ps_config['id']}",
+        *required_config_overrides,
+    ]
+    completed_init = run_pageserver(good_init_cmd)
+    assert (
+        completed_init.returncode == 0
+    ), "pageserver should be able to create a new config with the node id given"
+    assert pageserver_config.exists(), "config file should be created successfully"
+
+    bad_reinit = run_pageserver(good_init_cmd)
+    assert bad_reinit.returncode == 1, "pageserver refuses to init if already exists"
+    assert "config file already exists" in bad_reinit.stderr
+
+
 def check_client(env: NeonEnv, client: PageserverHttpClient):
     pg_version = env.pg_version
     initial_tenant = env.initial_tenant