[test_runner/performance] scale: Increase data volumes in PostgreSQL regression tests by 10x

Co-Authored-By: alexeymasterov@neon.tech <alexeymasterov@neon.tech>

Cargo.lock

@@ -4286,7 +4286,6 @@ dependencies = [
  "enumset",
  "fail",
  "futures",
- "hashlink",
  "hex",
  "hex-literal",
  "http-utils",

compute_tools/src/pg_helpers.rs

@@ -213,10 +213,8 @@ impl Escaping for PgIdent {
 
         // Find the first suitable tag that is not present in the string.
         // Postgres' max role/DB name length is 63 bytes, so even in the
-        // worst case it won't take long. Outer tag is always `tag + "x"`,
-        // so if `tag` is not present in the string, `outer_tag` is not
-        // present in the string either.
-        while self.contains(&tag.to_string()) {
+        // worst case it won't take long.
+        while self.contains(&format!("${tag}$")) || self.contains(&format!("${outer_tag}$")) {
             tag += "x";
             outer_tag = tag.clone() + "x";
         }

compute_tools/tests/pg_helpers_tests.rs

@@ -71,14 +71,6 @@ test.escaping = 'here''s a backslash \\ and a quote '' and a double-quote " hoor
             ("name$$$", ("$x$name$$$$x$", "xx")),
             ("name$$$$", ("$x$name$$$$$x$", "xx")),
             ("name$x$", ("$xx$name$x$$xx$", "xxx")),
-            ("x", ("$xx$x$xx$", "xxx")),
-            ("xx", ("$xxx$xx$xxx$", "xxxx")),
-            ("$x", ("$xx$$x$xx$", "xxx")),
-            ("x$", ("$xx$x$$xx$", "xxx")),
-            ("$x$", ("$xx$$x$$xx$", "xxx")),
-            ("xx$", ("$xxx$xx$$xxx$", "xxxx")),
-            ("$xx", ("$xxx$$xx$xxx$", "xxxx")),
-            ("$xx$", ("$xxx$$xx$$xxx$", "xxxx")),
         ];
 
         for (input, expected) in test_cases {

control_plane/src/pageserver.rs

@@ -546,11 +546,6 @@ impl PageServerNode {
                 .map(serde_json::from_str)
                 .transpose()
                 .context("Falied to parse 'sampling_ratio'")?,
-            relsize_snapshot_cache_capacity: settings
-                .remove("relsize snapshot cache capacity")
-                .map(|x| x.parse::<usize>())
-                .transpose()
-                .context("Falied to parse 'relsize_snapshot_cache_capacity' as integer")?,
         };
         if !settings.is_empty() {
             bail!("Unrecognized tenant settings: {settings:?}")

endpoint_storage/src/app.rs

@@ -462,8 +462,6 @@ MC4CAQAwBQYDK2VwBCIEID/Drmc1AA6U/znNRWpF3zEGegOATQxfkdWxitcOMsIH
         if var(REAL_S3_ENV).is_ok() {
             assert!(body.contains("remote_storage_s3_deleted_objects_total"));
         }
-
-        #[cfg(target_os = "linux")]
         assert!(body.contains("process_threads"));
     }
 

libs/pageserver_api/src/config.rs

@@ -235,7 +235,7 @@ pub enum PageServiceProtocolPipelinedBatchingStrategy {
     ScatteredLsn,
 }
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 #[serde(tag = "mode", rename_all = "kebab-case")]
 pub enum GetVectoredConcurrentIo {
     /// The read path is fully sequential: layers are visited
@@ -491,8 +491,6 @@ pub struct TenantConfigToml {
     /// Tenant level performance sampling ratio override. Controls the ratio of get page requests
     /// that will get perf sampling for the tenant.
     pub sampling_ratio: Option<Ratio>,
-    /// Capacity of relsize snapshot cache (used by replicas).
-    pub relsize_snapshot_cache_capacity: usize,
 }
 
 pub mod defaults {
@@ -732,7 +730,6 @@ pub mod tenant_conf_defaults {
     pub const DEFAULT_GC_COMPACTION_VERIFICATION: bool = true;
     pub const DEFAULT_GC_COMPACTION_INITIAL_THRESHOLD_KB: u64 = 5 * 1024 * 1024; // 5GB
     pub const DEFAULT_GC_COMPACTION_RATIO_PERCENT: u64 = 100;
-    pub const DEFAULT_RELSIZE_SNAPSHOT_CACHE_CAPACITY: usize = 1000;
 }
 
 impl Default for TenantConfigToml {
@@ -790,7 +787,6 @@ impl Default for TenantConfigToml {
             gc_compaction_initial_threshold_kb: DEFAULT_GC_COMPACTION_INITIAL_THRESHOLD_KB,
             gc_compaction_ratio_percent: DEFAULT_GC_COMPACTION_RATIO_PERCENT,
             sampling_ratio: None,
-            relsize_snapshot_cache_capacity: DEFAULT_RELSIZE_SNAPSHOT_CACHE_CAPACITY,
         }
     }
 }

libs/pageserver_api/src/models.rs

@@ -630,8 +630,6 @@ pub struct TenantConfigPatch {
     pub gc_compaction_ratio_percent: FieldPatch<u64>,
     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     pub sampling_ratio: FieldPatch<Option<Ratio>>,
-    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
-    pub relsize_snapshot_cache_capacity: FieldPatch<usize>,
 }
 
 /// Like [`crate::config::TenantConfigToml`], but preserves the information
@@ -761,9 +759,6 @@ pub struct TenantConfig {
 
     #[serde(skip_serializing_if = "Option::is_none")]
     pub sampling_ratio: Option<Option<Ratio>>,
-
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub relsize_snapshot_cache_capacity: Option<usize>,
 }
 
 impl TenantConfig {
@@ -809,7 +804,6 @@ impl TenantConfig {
             mut gc_compaction_initial_threshold_kb,
             mut gc_compaction_ratio_percent,
             mut sampling_ratio,
-            mut relsize_snapshot_cache_capacity,
         } = self;
 
         patch.checkpoint_distance.apply(&mut checkpoint_distance);
@@ -911,9 +905,6 @@ impl TenantConfig {
             .gc_compaction_ratio_percent
             .apply(&mut gc_compaction_ratio_percent);
         patch.sampling_ratio.apply(&mut sampling_ratio);
-        patch
-            .relsize_snapshot_cache_capacity
-            .apply(&mut relsize_snapshot_cache_capacity);
 
         Ok(Self {
             checkpoint_distance,
@@ -953,7 +944,6 @@ impl TenantConfig {
             gc_compaction_initial_threshold_kb,
             gc_compaction_ratio_percent,
             sampling_ratio,
-            relsize_snapshot_cache_capacity,
         })
     }
 
@@ -1062,9 +1052,6 @@ impl TenantConfig {
                 .gc_compaction_ratio_percent
                 .unwrap_or(global_conf.gc_compaction_ratio_percent),
             sampling_ratio: self.sampling_ratio.unwrap_or(global_conf.sampling_ratio),
-            relsize_snapshot_cache_capacity: self
-                .relsize_snapshot_cache_capacity
-                .unwrap_or(global_conf.relsize_snapshot_cache_capacity),
         }
     }
 }

libs/proxy/postgres-protocol2/src/authentication/sasl.rs

@@ -9,14 +9,12 @@ use sha2::digest::FixedOutput;
 use sha2::{Digest, Sha256};
 use tokio::task::yield_now;
 
-use crate::CSafeStr;
-
 const NONCE_LENGTH: usize = 24;
 
 /// The identifier of the SCRAM-SHA-256 SASL authentication mechanism.
-pub const SCRAM_SHA_256: &CSafeStr = CSafeStr::from_cstr(c"SCRAM-SHA-256");
+pub const SCRAM_SHA_256: &str = "SCRAM-SHA-256";
 /// The identifier of the SCRAM-SHA-256-PLUS SASL authentication mechanism.
-pub const SCRAM_SHA_256_PLUS: &CSafeStr = CSafeStr::from_cstr(c"SCRAM-SHA-256-PLUS");
+pub const SCRAM_SHA_256_PLUS: &str = "SCRAM-SHA-256-PLUS";
 
 // since postgres passwords are not required to exclude saslprep-prohibited
 // characters or even be valid UTF8, we run saslprep if possible and otherwise

libs/proxy/postgres-protocol2/src/lib.rs

@@ -11,7 +11,7 @@
 //! set to `UTF8`. It will most likely not behave properly if that is not the case.
 #![warn(missing_docs, clippy::all)]
 
-use std::{ffi::CStr, io};
+use std::io;
 
 use byteorder::{BigEndian, ByteOrder};
 use bytes::{BufMut, BytesMut};
@@ -36,67 +36,20 @@ pub enum IsNull {
     No,
 }
 
-/// A [`std::ffi::CStr`] but without the null byte.
-#[repr(transparent)]
-#[derive(PartialEq, Eq, Hash, Debug)]
-pub struct CSafeStr([u8]);
-
-impl CSafeStr {
-    /// Create a new `CSafeStr`, erroring if the bytes contains a null.
-    pub fn new(bytes: &[u8]) -> Result<&Self, io::Error> {
-        let nul_pos = memchr::memchr(0, bytes);
-        match nul_pos {
-            Some(nul_pos) => Err(io::Error::other(format!(
-                "unexpected null byte at position {nul_pos}"
-            ))),
-            None => {
-                // Safety: CSafeStr is transparent over [u8].
-                Ok(unsafe { std::mem::transmute(bytes) })
-            }
-        }
-    }
-
-    /// Create a new `CSafeStr` up until the next null.
-    pub fn take<'a>(bytes: &mut &'a [u8]) -> &'a Self {
-        let nul_pos = memchr::memchr(0, bytes).unwrap_or(bytes.len());
-        let bytes = bytes
-            .split_off(..nul_pos)
-            .expect("nul_pos should be in-bounds");
-        // Safety: CSafeStr is transparent over [u8].
-        unsafe { std::mem::transmute(bytes) }
-    }
-
-    /// Get the bytes of this CSafeStr.
-    pub const fn as_bytes(&self) -> &[u8] {
-        &self.0
-    }
-
-    /// Create a new `CSafeStr`
-    pub const fn from_cstr(s: &CStr) -> &CSafeStr {
-        // Safety: CSafeStr is transparent over [u8].
-        unsafe { std::mem::transmute(s.to_bytes()) }
-    }
-}
-
-impl<'a> From<&'a CStr> for &'a CSafeStr {
-    fn from(s: &'a CStr) -> &'a CSafeStr {
-        CSafeStr::from_cstr(s)
-    }
-}
-
-fn write_nullable<F>(serializer: F, buf: &mut BytesMut)
+fn write_nullable<F, E>(serializer: F, buf: &mut BytesMut) -> Result<(), E>
 where
-    F: FnOnce(&mut BytesMut) -> IsNull,
+    F: FnOnce(&mut BytesMut) -> Result<IsNull, E>,
+    E: From<io::Error>,
 {
     let base = buf.len();
     buf.put_i32(0);
-    let size = match serializer(buf) {
-        // this is an unreasonable enough case that I think a panic is acceptable.
-        IsNull::No => i32::from_usize(buf.len() - base - 4)
-            .expect("buffer size should not be larger than i32::MAX"),
+    let size = match serializer(buf)? {
+        IsNull::No => i32::from_usize(buf.len() - base - 4)?,
         IsNull::Yes => -1,
     };
     BigEndian::write_i32(&mut buf[base..], size);
+
+    Ok(())
 }
 
 trait FromUsize: Sized {

libs/proxy/postgres-protocol2/src/message/backend.rs

@@ -9,7 +9,7 @@ use bytes::{Bytes, BytesMut};
 use fallible_iterator::FallibleIterator;
 use memchr::memchr;
 
-use crate::{CSafeStr, Oid};
+use crate::Oid;
 
 // top-level message tags
 const PARSE_COMPLETE_TAG: u8 = b'1';
@@ -332,24 +332,25 @@ impl AuthenticationSaslBody {
 pub struct SaslMechanisms<'a>(&'a [u8]);
 
 impl<'a> FallibleIterator for SaslMechanisms<'a> {
-    type Item = &'a CSafeStr;
+    type Item = &'a str;
     type Error = io::Error;
 
     #[inline]
-    fn next(&mut self) -> io::Result<Option<&'a CSafeStr>> {
-        if self.0 == b"0" {
-            return Ok(None);
+    fn next(&mut self) -> io::Result<Option<&'a str>> {
+        let value_end = find_null(self.0, 0)?;
+        if value_end == 0 {
+            if self.0.len() != 1 {
+                return Err(io::Error::new(
+                    io::ErrorKind::InvalidData,
+                    "invalid message length: expected to be at end of iterator for sasl",
+                ));
+            }
+            Ok(None)
+        } else {
+            let value = get_str(&self.0[..value_end])?;
+            self.0 = &self.0[value_end + 1..];
+            Ok(Some(value))
         }
-
-        let value = CSafeStr::take(&mut self.0);
-        if value.as_bytes().len() == 0 {
-            return Err(io::Error::new(
-                io::ErrorKind::InvalidData,
-                "invalid message length: expected to be at end of iterator for sasl",
-            ));
-        }
-
-        Ok(Some(value))
     }
 }
 

libs/proxy/postgres-protocol2/src/message/frontend.rs

@@ -1,73 +1,114 @@
 //! Frontend message serialization.
 #![allow(missing_docs)]
 
-use std::io;
+use std::error::Error;
+use std::{io, marker};
 
 use byteorder::{BigEndian, ByteOrder};
 use bytes::{Buf, BufMut, BytesMut};
 
-use crate::{CSafeStr, FromUsize, IsNull, Oid, write_nullable};
+use crate::{FromUsize, IsNull, Oid, write_nullable};
 
 #[inline]
-fn write_body<F>(buf: &mut BytesMut, f: F)
+fn write_body<F, E>(buf: &mut BytesMut, f: F) -> Result<(), E>
 where
-    F: FnOnce(&mut BytesMut),
+    F: FnOnce(&mut BytesMut) -> Result<(), E>,
+    E: From<io::Error>,
 {
     let base = buf.len();
     buf.extend_from_slice(&[0; 4]);
 
-    f(buf);
+    f(buf)?;
 
-    let size =
-        i32::from_usize(buf.len() - base).expect("buffer size should not be larger than i32::MAX");
+    let size = i32::from_usize(buf.len() - base)?;
     BigEndian::write_i32(&mut buf[base..], size);
+    Ok(())
+}
+
+pub enum BindError {
+    Conversion(Box<dyn Error + marker::Sync + Send>),
+    Serialization(io::Error),
+}
+
+impl From<Box<dyn Error + marker::Sync + Send>> for BindError {
+    #[inline]
+    fn from(e: Box<dyn Error + marker::Sync + Send>) -> BindError {
+        BindError::Conversion(e)
+    }
+}
+
+impl From<io::Error> for BindError {
+    #[inline]
+    fn from(e: io::Error) -> BindError {
+        BindError::Serialization(e)
+    }
 }
 
 #[inline]
 pub fn bind<I, J, F, T, K>(
-    portal: &CSafeStr,
-    statement: &CSafeStr,
+    portal: &str,
+    statement: &str,
     formats: I,
     values: J,
     mut serializer: F,
     result_formats: K,
     buf: &mut BytesMut,
-) where
+) -> Result<(), BindError>
+where
     I: IntoIterator<Item = i16>,
     J: IntoIterator<Item = T>,
-    F: FnMut(T, &mut BytesMut) -> IsNull,
+    F: FnMut(T, &mut BytesMut) -> Result<IsNull, Box<dyn Error + marker::Sync + Send>>,
     K: IntoIterator<Item = i16>,
 {
     buf.put_u8(b'B');
 
     write_body(buf, |buf| {
-        write_cstr(portal, buf);
-        write_cstr(statement, buf);
-        write_counted(formats, |f, buf| buf.put_i16(f), buf);
+        write_cstr(portal.as_bytes(), buf)?;
+        write_cstr(statement.as_bytes(), buf)?;
+        write_counted(
+            formats,
+            |f, buf| {
+                buf.put_i16(f);
+                Ok::<_, io::Error>(())
+            },
+            buf,
+        )?;
         write_counted(
             values,
             |v, buf| write_nullable(|buf| serializer(v, buf), buf),
             buf,
-        );
-        write_counted(result_formats, |f, buf| buf.put_i16(f), buf);
+        )?;
+        write_counted(
+            result_formats,
+            |f, buf| {
+                buf.put_i16(f);
+                Ok::<_, io::Error>(())
+            },
+            buf,
+        )?;
+
+        Ok(())
     })
 }
 
 #[inline]
-fn write_counted<I, T, F>(items: I, mut serializer: F, buf: &mut BytesMut)
+fn write_counted<I, T, F, E>(items: I, mut serializer: F, buf: &mut BytesMut) -> Result<(), E>
 where
     I: IntoIterator<Item = T>,
-    F: FnMut(T, &mut BytesMut),
+    F: FnMut(T, &mut BytesMut) -> Result<(), E>,
+    E: From<io::Error>,
 {
     let base = buf.len();
     buf.extend_from_slice(&[0; 2]);
     let mut count = 0;
     for item in items {
-        serializer(item, buf);
+        serializer(item, buf)?;
         count += 1;
     }
-    let count = i16::from_usize(count).expect("list should not exceed 32767 items");
+    let count = i16::from_usize(count)?;
     BigEndian::write_i16(&mut buf[base..], count);
+
+    Ok(())
 }
 
 #[inline]
@@ -76,15 +117,17 @@ pub fn cancel_request(process_id: i32, secret_key: i32, buf: &mut BytesMut) {
         buf.put_i32(80_877_102);
         buf.put_i32(process_id);
         buf.put_i32(secret_key);
+        Ok::<_, io::Error>(())
     })
+    .unwrap();
 }
 
 #[inline]
-pub fn close(variant: u8, name: &CSafeStr, buf: &mut BytesMut) {
+pub fn close(variant: u8, name: &str, buf: &mut BytesMut) -> io::Result<()> {
     buf.put_u8(b'C');
     write_body(buf, |buf| {
         buf.put_u8(variant);
-        write_cstr(name, buf)
+        write_cstr(name.as_bytes(), buf)
     })
 }
 
@@ -119,75 +162,85 @@ where
 #[inline]
 pub fn copy_done(buf: &mut BytesMut) {
     buf.put_u8(b'c');
-    write_body(buf, |_| {});
+    write_body(buf, |_| Ok::<(), io::Error>(())).unwrap();
 }
 
 #[inline]
-pub fn copy_fail(message: &CSafeStr, buf: &mut BytesMut) {
+pub fn copy_fail(message: &str, buf: &mut BytesMut) -> io::Result<()> {
     buf.put_u8(b'f');
-    write_body(buf, |buf| write_cstr(message, buf))
+    write_body(buf, |buf| write_cstr(message.as_bytes(), buf))
 }
 
 #[inline]
-pub fn describe(variant: u8, name: &CSafeStr, buf: &mut BytesMut) {
+pub fn describe(variant: u8, name: &str, buf: &mut BytesMut) -> io::Result<()> {
     buf.put_u8(b'D');
     write_body(buf, |buf| {
         buf.put_u8(variant);
-        write_cstr(name, buf)
+        write_cstr(name.as_bytes(), buf)
     })
 }
 
 #[inline]
-pub fn execute(portal: &CSafeStr, max_rows: i32, buf: &mut BytesMut) {
+pub fn execute(portal: &str, max_rows: i32, buf: &mut BytesMut) -> io::Result<()> {
     buf.put_u8(b'E');
     write_body(buf, |buf| {
-        write_cstr(portal, buf);
+        write_cstr(portal.as_bytes(), buf)?;
         buf.put_i32(max_rows);
+        Ok(())
     })
 }
 
 #[inline]
-pub fn parse<I>(name: &CSafeStr, query: &CSafeStr, param_types: I, buf: &mut BytesMut)
+pub fn parse<I>(name: &str, query: &str, param_types: I, buf: &mut BytesMut) -> io::Result<()>
 where
     I: IntoIterator<Item = Oid>,
 {
     buf.put_u8(b'P');
     write_body(buf, |buf| {
-        write_cstr(name, buf);
-        write_cstr(query, buf);
-        write_counted(param_types, |t, buf| buf.put_u32(t), buf);
+        write_cstr(name.as_bytes(), buf)?;
+        write_cstr(query.as_bytes(), buf)?;
+        write_counted(
+            param_types,
+            |t, buf| {
+                buf.put_u32(t);
+                Ok::<_, io::Error>(())
+            },
+            buf,
+        )?;
+        Ok(())
     })
 }
 
 #[inline]
-pub fn password_message(password: &CSafeStr, buf: &mut BytesMut) {
+pub fn password_message(password: &[u8], buf: &mut BytesMut) -> io::Result<()> {
     buf.put_u8(b'p');
     write_body(buf, |buf| write_cstr(password, buf))
 }
 
 #[inline]
-pub fn query(query: &CSafeStr, buf: &mut BytesMut) {
+pub fn query(query: &str, buf: &mut BytesMut) -> io::Result<()> {
     buf.put_u8(b'Q');
-    write_body(buf, |buf| write_cstr(query, buf))
+    write_body(buf, |buf| write_cstr(query.as_bytes(), buf))
 }
 
 #[inline]
-pub fn sasl_initial_response(mechanism: &CSafeStr, data: &[u8], buf: &mut BytesMut) {
+pub fn sasl_initial_response(mechanism: &str, data: &[u8], buf: &mut BytesMut) -> io::Result<()> {
     buf.put_u8(b'p');
     write_body(buf, |buf| {
-        write_cstr(mechanism, buf);
-        let len =
-            i32::from_usize(data.len()).expect("sasl data should not be larger than i32::MAX");
+        write_cstr(mechanism.as_bytes(), buf)?;
+        let len = i32::from_usize(data.len())?;
         buf.put_i32(len);
         buf.put_slice(data);
+        Ok(())
     })
 }
 
 #[inline]
-pub fn sasl_response(data: &[u8], buf: &mut BytesMut) {
+pub fn sasl_response(data: &[u8], buf: &mut BytesMut) -> io::Result<()> {
     buf.put_u8(b'p');
     write_body(buf, |buf| {
         buf.put_slice(data);
+        Ok(())
     })
 }
 
@@ -195,16 +248,19 @@ pub fn sasl_response(data: &[u8], buf: &mut BytesMut) {
 pub fn ssl_request(buf: &mut BytesMut) {
     write_body(buf, |buf| {
         buf.put_i32(80_877_103);
-    });
+        Ok::<_, io::Error>(())
+    })
+    .unwrap();
 }
 
 #[inline]
-pub fn startup_message(parameters: &StartupMessageParams, buf: &mut BytesMut) {
+pub fn startup_message(parameters: &StartupMessageParams, buf: &mut BytesMut) -> io::Result<()> {
     write_body(buf, |buf| {
         // postgres protocol version 3.0(196608) in bigger-endian
         buf.put_i32(0x00_03_00_00);
         buf.put_slice(&parameters.params);
         buf.put_u8(0);
+        Ok(())
     })
 }
 
@@ -215,7 +271,10 @@ pub struct StartupMessageParams {
 
 impl StartupMessageParams {
     /// Set parameter's value by its name.
-    pub fn insert(&mut self, name: &CSafeStr, value: &CSafeStr) {
+    pub fn insert(&mut self, name: &str, value: &str) {
+        if name.contains('\0') || value.contains('\0') {
+            panic!("startup parameter name or value contained a null")
+        }
         self.params.put_slice(name.as_bytes());
         self.params.put_u8(0);
         self.params.put_slice(value.as_bytes());
@@ -226,17 +285,24 @@ impl StartupMessageParams {
 #[inline]
 pub fn sync(buf: &mut BytesMut) {
     buf.put_u8(b'S');
-    write_body(buf, |_| {});
+    write_body(buf, |_| Ok::<(), io::Error>(())).unwrap();
 }
 
 #[inline]
 pub fn terminate(buf: &mut BytesMut) {
     buf.put_u8(b'X');
-    write_body(buf, |_| {});
+    write_body(buf, |_| Ok::<(), io::Error>(())).unwrap();
 }
 
 #[inline]
-fn write_cstr(s: &CSafeStr, buf: &mut BytesMut) {
-    buf.put_slice(s.as_bytes());
+fn write_cstr(s: &[u8], buf: &mut BytesMut) -> Result<(), io::Error> {
+    if s.contains(&0) {
+        return Err(io::Error::new(
+            io::ErrorKind::InvalidInput,
+            "string contains embedded null",
+        ));
+    }
+    buf.put_slice(s);
     buf.put_u8(0);
+    Ok(())
 }

libs/proxy/postgres-types2/src/lib.rs

@@ -13,7 +13,7 @@ use bytes::BytesMut;
 use fallible_iterator::FallibleIterator;
 #[doc(inline)]
 pub use postgres_protocol2::Oid;
-use postgres_protocol2::{IsNull, types};
+use postgres_protocol2::types;
 
 use crate::type_gen::{Inner, Other};
 
@@ -32,15 +32,36 @@ macro_rules! accepts {
 /// All `ToSql` implementations should use this macro.
 macro_rules! to_sql_checked {
     () => {
-        fn check(&self, ty: &Type) -> ::std::result::Result<(), $crate::WrongType> {
-            if !<Self as $crate::ToSql>::accepts(ty) {
-                return Err($crate::WrongType::new::<Self>(ty.clone()));
-            }
-            Ok(())
+        fn to_sql_checked(
+            &self,
+            ty: &$crate::Type,
+            out: &mut $crate::private::BytesMut,
+        ) -> ::std::result::Result<
+            $crate::IsNull,
+            Box<dyn ::std::error::Error + ::std::marker::Sync + ::std::marker::Send>,
+        > {
+            $crate::__to_sql_checked(self, ty, out)
         }
     };
 }
 
+// WARNING: this function is not considered part of this crate's public API.
+// It is subject to change at any time.
+#[doc(hidden)]
+pub fn __to_sql_checked<T>(
+    v: &T,
+    ty: &Type,
+    out: &mut BytesMut,
+) -> Result<IsNull, Box<dyn Error + Sync + Send>>
+where
+    T: ToSql,
+{
+    if !T::accepts(ty) {
+        return Err(Box::new(WrongType::new::<T>(ty.clone())));
+    }
+    v.to_sql(ty, out)
+}
+
 // mod pg_lsn;
 #[doc(hidden)]
 pub mod private;
@@ -348,6 +369,14 @@ macro_rules! simple_from {
 simple_from!(i8, char_from_sql, CHAR);
 simple_from!(u32, oid_from_sql, OID);
 
+/// An enum representing the nullability of a Postgres value.
+pub enum IsNull {
+    /// The value is NULL.
+    Yes,
+    /// The value is not NULL.
+    No,
+}
+
 /// A trait for types that can be converted into Postgres values.
 pub trait ToSql: fmt::Debug {
     /// Converts the value of `self` into the binary format of the specified
@@ -359,7 +388,9 @@ pub trait ToSql: fmt::Debug {
     /// The return value indicates if this value should be represented as
     /// `NULL`. If this is the case, implementations **must not** write
     /// anything to `out`.
-    fn to_sql(&self, ty: &Type, out: &mut BytesMut) -> IsNull;
+    fn to_sql(&self, ty: &Type, out: &mut BytesMut) -> Result<IsNull, Box<dyn Error + Sync + Send>>
+    where
+        Self: Sized;
 
     /// Determines if a value of this type can be converted to the specified
     /// Postgres `Type`.
@@ -371,7 +402,11 @@ pub trait ToSql: fmt::Debug {
     ///
     /// *All* implementations of this method should be generated by the
     /// `to_sql_checked!()` macro.
-    fn check(&self, ty: &Type) -> Result<(), WrongType>;
+    fn to_sql_checked(
+        &self,
+        ty: &Type,
+        out: &mut BytesMut,
+    ) -> Result<IsNull, Box<dyn Error + Sync + Send>>;
 
     /// Specify the encode format
     fn encode_format(&self, _ty: &Type) -> Format {
@@ -391,14 +426,14 @@ pub enum Format {
 }
 
 impl ToSql for &str {
-    fn to_sql(&self, ty: &Type, w: &mut BytesMut) -> IsNull {
+    fn to_sql(&self, ty: &Type, w: &mut BytesMut) -> Result<IsNull, Box<dyn Error + Sync + Send>> {
         match *ty {
             ref ty if ty.name() == "ltree" => types::ltree_to_sql(self, w),
             ref ty if ty.name() == "lquery" => types::lquery_to_sql(self, w),
             ref ty if ty.name() == "ltxtquery" => types::ltxtquery_to_sql(self, w),
             _ => types::text_to_sql(self, w),
         }
-        IsNull::No
+        Ok(IsNull::No)
     }
 
     fn accepts(ty: &Type) -> bool {
@@ -422,9 +457,12 @@ impl ToSql for &str {
 macro_rules! simple_to {
     ($t:ty, $f:ident, $($expected:ident),+) => {
         impl ToSql for $t {
-            fn to_sql(&self, _: &Type, w: &mut BytesMut) -> IsNull {
+            fn to_sql(&self,
+                      _: &Type,
+                      w: &mut BytesMut)
+                      -> Result<IsNull, Box<dyn Error + Sync + Send>> {
                 types::$f(*self, w);
-                IsNull::No
+                Ok(IsNull::No)
             }
 
             accepts!($($expected),+);

libs/proxy/tokio-postgres2/src/client.rs

@@ -219,7 +219,7 @@ impl Client {
                 }
 
                 let buf = self.client.inner().with_buf(|buf| {
-                    frontend::query(c"ROLLBACK".into(), buf);
+                    frontend::query("ROLLBACK", buf).unwrap();
                     buf.split().freeze()
                 });
                 let _ = self

libs/proxy/tokio-postgres2/src/config.rs

@@ -4,7 +4,6 @@ use std::net::IpAddr;
 use std::time::Duration;
 use std::{fmt, str};
 
-use postgres_protocol2::CSafeStr;
 pub use postgres_protocol2::authentication::sasl::ScramKeys;
 use postgres_protocol2::message::frontend::StartupMessageParams;
 use serde::{Deserialize, Serialize};
@@ -163,10 +162,7 @@ impl Config {
             self.username = true;
         }
 
-        self.server_params.insert(
-            CSafeStr::new(name.as_bytes()).expect("param name should not contain a null"),
-            CSafeStr::new(value.as_bytes()).expect("param name should not contain a null"),
-        );
+        self.server_params.insert(name, value);
         self
     }
 

libs/proxy/tokio-postgres2/src/connect_raw.rs

@@ -6,7 +6,6 @@ use std::task::{Context, Poll};
 use bytes::BytesMut;
 use fallible_iterator::FallibleIterator;
 use futures_util::{Sink, SinkExt, Stream, TryStreamExt, ready};
-use postgres_protocol2::CSafeStr;
 use postgres_protocol2::authentication::sasl;
 use postgres_protocol2::authentication::sasl::ScramSha256;
 use postgres_protocol2::message::backend::{AuthenticationSaslBody, Message, NoticeResponseBody};
@@ -123,7 +122,7 @@ where
     T: AsyncRead + AsyncWrite + Unpin,
 {
     let mut buf = BytesMut::new();
-    frontend::startup_message(&config.server_params, &mut buf);
+    frontend::startup_message(&config.server_params, &mut buf).map_err(Error::encode)?;
 
     stream
         .send(FrontendMessage::Raw(buf.freeze()))
@@ -194,7 +193,7 @@ where
     T: AsyncRead + AsyncWrite + Unpin,
 {
     let mut buf = BytesMut::new();
-    frontend::password_message(CSafeStr::new(password).map_err(Error::encode)?, &mut buf);
+    frontend::password_message(password, &mut buf).map_err(Error::encode)?;
 
     stream
         .send(FrontendMessage::Raw(buf.freeze()))
@@ -215,10 +214,10 @@ where
     let mut has_scram_plus = false;
     let mut mechanisms = body.mechanisms();
     while let Some(mechanism) = mechanisms.next().map_err(Error::parse)? {
-        if mechanism == sasl::SCRAM_SHA_256 {
-            has_scram = true;
-        } else if mechanism == sasl::SCRAM_SHA_256_PLUS {
-            has_scram_plus = true;
+        match mechanism {
+            sasl::SCRAM_SHA_256 => has_scram = true,
+            sasl::SCRAM_SHA_256_PLUS => has_scram_plus = true,
+            _ => {}
         }
     }
 
@@ -257,7 +256,7 @@ where
     };
 
     let mut buf = BytesMut::new();
-    frontend::sasl_initial_response(mechanism, scram.message(), &mut buf);
+    frontend::sasl_initial_response(mechanism, scram.message(), &mut buf).map_err(Error::encode)?;
     stream
         .send(FrontendMessage::Raw(buf.freeze()))
         .await
@@ -276,7 +275,7 @@ where
         .map_err(|e| Error::authentication(e.into()))?;
 
     let mut buf = BytesMut::new();
-    frontend::sasl_response(scram.message(), &mut buf);
+    frontend::sasl_response(scram.message(), &mut buf).map_err(Error::encode)?;
     stream
         .send(FrontendMessage::Raw(buf.freeze()))
         .await

libs/proxy/tokio-postgres2/src/lib.rs

@@ -123,6 +123,6 @@ pub enum SimpleQueryMessage {
 
 fn slice_iter<'a>(
     s: &'a [&'a (dyn ToSql + Sync)],
-) -> impl ExactSizeIterator<Item = &'a (dyn ToSql + Sync)> + Clone + 'a {
+) -> impl ExactSizeIterator<Item = &'a (dyn ToSql + Sync)> + 'a {
     s.iter().map(|s| *s as _)
 }

libs/proxy/tokio-postgres2/src/prepare.rs

@@ -1,4 +1,3 @@
-use std::ffi::CStr;
 use std::future::Future;
 use std::pin::Pin;
 use std::sync::Arc;
@@ -6,9 +5,9 @@ use std::sync::Arc;
 use bytes::Bytes;
 use fallible_iterator::FallibleIterator;
 use futures_util::{TryStreamExt, pin_mut};
-use postgres_protocol2::CSafeStr;
 use postgres_protocol2::message::backend::Message;
 use postgres_protocol2::message::frontend;
+use tracing::debug;
 
 use crate::client::{CachedTypeInfo, InnerClient};
 use crate::codec::FrontendMessage;
@@ -16,7 +15,7 @@ use crate::connection::RequestMessages;
 use crate::types::{Kind, Oid, Type};
 use crate::{Column, Error, Statement, query, slice_iter};
 
-pub(crate) const TYPEINFO_QUERY: &CStr = c"\
+pub(crate) const TYPEINFO_QUERY: &str = "\
 SELECT t.typname, t.typtype, t.typelem, r.rngsubtype, t.typbasetype, n.nspname, t.typrelid
 FROM pg_catalog.pg_type t
 LEFT OUTER JOIN pg_catalog.pg_range r ON r.rngtypid = t.oid
@@ -26,11 +25,11 @@ WHERE t.oid = $1
 
 async fn prepare_typecheck(
     client: &Arc<InnerClient>,
-    name: &'static CStr,
-    query: &CSafeStr,
+    name: &'static str,
+    query: &str,
     types: &[Type],
 ) -> Result<Statement, Error> {
-    let buf = encode(client, name.into(), query, types)?;
+    let buf = encode(client, name, query, types)?;
     let mut responses = client.send(RequestMessages::Single(FrontendMessage::Raw(buf)))?;
 
     match responses.next().await? {
@@ -69,21 +68,16 @@ async fn prepare_typecheck(
     Ok(Statement::new(client, name, parameters, columns))
 }
 
-fn encode(
-    client: &InnerClient,
-    name: &CSafeStr,
-    query: &CSafeStr,
-    types: &[Type],
-) -> Result<Bytes, Error> {
-    // if types.is_empty() {
-    //     debug!("preparing query {}: {}", name, query);
-    // } else {
-    //     debug!("preparing query {} with types {:?}: {}", name, types, query);
-    // }
+fn encode(client: &InnerClient, name: &str, query: &str, types: &[Type]) -> Result<Bytes, Error> {
+    if types.is_empty() {
+        debug!("preparing query {}: {}", name, query);
+    } else {
+        debug!("preparing query {} with types {:?}: {}", name, types, query);
+    }
 
     client.with_buf(|buf| {
-        frontend::parse(name, query, types.iter().map(Type::oid), buf);
-        frontend::describe(b'S', name, buf);
+        frontend::parse(name, query, types.iter().map(Type::oid), buf).map_err(Error::encode)?;
+        frontend::describe(b'S', name, buf).map_err(Error::encode)?;
         frontend::sync(buf);
         Ok(buf.split().freeze())
     })
@@ -160,8 +154,8 @@ async fn typeinfo_statement(
         return Ok(stmt.clone());
     }
 
-    let typeinfo = c"neon_proxy_typeinfo";
-    let stmt = prepare_typecheck(client, typeinfo, TYPEINFO_QUERY.into(), &[]).await?;
+    let typeinfo = "neon_proxy_typeinfo";
+    let stmt = prepare_typecheck(client, typeinfo, TYPEINFO_QUERY, &[]).await?;
 
     typecache.typeinfo = Some(stmt.clone());
     Ok(stmt)

libs/proxy/tokio-postgres2/src/query.rs

@@ -1,4 +1,4 @@
-use std::iter;
+use std::fmt;
 use std::marker::PhantomPinned;
 use std::pin::Pin;
 use std::sync::Arc;
@@ -8,16 +8,25 @@ use bytes::{BufMut, Bytes, BytesMut};
 use fallible_iterator::FallibleIterator;
 use futures_util::{Stream, ready};
 use pin_project_lite::pin_project;
-use postgres_protocol2::CSafeStr;
 use postgres_protocol2::message::backend::Message;
 use postgres_protocol2::message::frontend;
 use postgres_types2::{Format, ToSql, Type};
+use tracing::debug;
 
 use crate::client::{InnerClient, Responses};
 use crate::codec::FrontendMessage;
 use crate::connection::RequestMessages;
+use crate::types::IsNull;
 use crate::{Column, Error, ReadyForQueryStatus, Row, Statement};
 
+struct BorrowToSqlParamsDebug<'a>(&'a [&'a (dyn ToSql + Sync)]);
+
+impl fmt::Debug for BorrowToSqlParamsDebug<'_> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_list().entries(self.0.iter()).finish()
+    }
+}
+
 pub async fn query<'a, I>(
     client: &InnerClient,
     statement: Statement,
@@ -25,9 +34,19 @@ pub async fn query<'a, I>(
 ) -> Result<RowStream, Error>
 where
     I: IntoIterator<Item = &'a (dyn ToSql + Sync)>,
-    I::IntoIter: ExactSizeIterator + Clone,
+    I::IntoIter: ExactSizeIterator,
 {
-    let buf = encode(client, &statement, params)?;
+    let buf = if tracing::enabled!(tracing::Level::DEBUG) {
+        let params = params.into_iter().collect::<Vec<_>>();
+        debug!(
+            "executing statement {} with parameters: {:?}",
+            statement.name(),
+            BorrowToSqlParamsDebug(params.as_slice()),
+        );
+        encode(client, &statement, params)?
+    } else {
+        encode(client, &statement, params)?
+    };
     let responses = start(client, buf).await?;
     Ok(RowStream {
         statement,
@@ -49,45 +68,45 @@ where
     I: IntoIterator<Item = Option<S>>,
     I::IntoIter: ExactSizeIterator,
 {
-    let query = CSafeStr::new(query.as_bytes()).map_err(Error::encode)?;
     let params = params.into_iter();
 
-    let portal = c"".into(); // unnamed portal
-    let statement = c"".into(); // unnamed prepared statement
-
     let buf = client.with_buf(|buf| {
         frontend::parse(
-            statement,
-            query,         // query to parse
-            iter::empty(), // give no type info
+            "",                 // unnamed prepared statement
+            query,              // query to parse
+            std::iter::empty(), // give no type info
             buf,
-        );
-        frontend::describe(b'S', statement, buf);
-
-        // Bind, pass params as text, retrieve as test
-        frontend::bind(
-            portal,
-            statement,
-            iter::empty(), // all parameters use the default format (text)
+        )
+        .map_err(Error::encode)?;
+        frontend::describe(b'S', "", buf).map_err(Error::encode)?;
+        // Bind, pass params as text, retrieve as binary
+        match frontend::bind(
+            "",                 // empty string selects the unnamed portal
+            "",                 // unnamed prepared statement
+            std::iter::empty(), // all parameters use the default format (text)
             params,
             |param, buf| match param {
                 Some(param) => {
                     buf.put_slice(param.as_ref().as_bytes());
-                    postgres_protocol2::IsNull::No
+                    Ok(postgres_protocol2::IsNull::No)
                 }
-                None => postgres_protocol2::IsNull::Yes,
+                None => Ok(postgres_protocol2::IsNull::Yes),
             },
             Some(0), // all text
             buf,
-        );
+        ) {
+            Ok(()) => Ok(()),
+            Err(frontend::BindError::Conversion(e)) => Err(Error::to_sql(e, 0)),
+            Err(frontend::BindError::Serialization(e)) => Err(Error::encode(e)),
+        }?;
 
         // Execute
-        frontend::execute(portal, 0, buf);
+        frontend::execute("", 0, buf).map_err(Error::encode)?;
         // Sync
         frontend::sync(buf);
 
-        buf.split().freeze()
-    });
+        Ok(buf.split().freeze())
+    })?;
 
     // now read the responses
     let mut responses = client.send(RequestMessages::Single(FrontendMessage::Raw(buf)))?;
@@ -154,13 +173,11 @@ async fn start(client: &InnerClient, buf: Bytes) -> Result<Responses, Error> {
 pub fn encode<'a, I>(client: &InnerClient, statement: &Statement, params: I) -> Result<Bytes, Error>
 where
     I: IntoIterator<Item = &'a (dyn ToSql + Sync)>,
-    I::IntoIter: ExactSizeIterator + Clone,
+    I::IntoIter: ExactSizeIterator,
 {
-    let portal = c"".into(); // unnamed portal
-
     client.with_buf(|buf| {
-        encode_bind(statement, params, portal, buf)?;
-        frontend::execute(portal, 0, buf);
+        encode_bind(statement, params, "", buf)?;
+        frontend::execute("", 0, buf).map_err(Error::encode)?;
         frontend::sync(buf);
         Ok(buf.split().freeze())
     })
@@ -169,15 +186,15 @@ where
 pub fn encode_bind<'a, I>(
     statement: &Statement,
     params: I,
-    portal: &CSafeStr,
+    portal: &str,
     buf: &mut BytesMut,
 ) -> Result<(), Error>
 where
     I: IntoIterator<Item = &'a (dyn ToSql + Sync)>,
-    I::IntoIter: ExactSizeIterator + Clone,
+    I::IntoIter: ExactSizeIterator,
 {
     let param_types = statement.params();
-    let params = iter::zip(params.into_iter(), param_types);
+    let params = params.into_iter();
 
     assert!(
         param_types.len() == params.len(),
@@ -186,24 +203,35 @@ where
         params.len()
     );
 
-    // check encodings
-    for (i, (p, ty)) in params.clone().enumerate() {
-        p.check(ty).map_err(|e| Error::to_sql(Box::new(e), i))?
-    }
+    let (param_formats, params): (Vec<_>, Vec<_>) = params
+        .zip(param_types.iter())
+        .map(|(p, ty)| (p.encode_format(ty) as i16, p))
+        .unzip();
 
-    let param_formats = params.clone().map(|(p, ty)| p.encode_format(ty) as i16);
+    let params = params.into_iter();
 
-    frontend::bind(
+    let mut error_idx = 0;
+    let r = frontend::bind(
         portal,
         statement.name(),
         param_formats,
-        params,
-        |(param, ty), buf| param.to_sql(ty, buf),
+        params.zip(param_types).enumerate(),
+        |(idx, (param, ty)), buf| match param.to_sql_checked(ty, buf) {
+            Ok(IsNull::No) => Ok(postgres_protocol2::IsNull::No),
+            Ok(IsNull::Yes) => Ok(postgres_protocol2::IsNull::Yes),
+            Err(e) => {
+                error_idx = idx;
+                Err(e)
+            }
+        },
         Some(1),
         buf,
     );
-
-    Ok(())
+    match r {
+        Ok(()) => Ok(()),
+        Err(frontend::BindError::Conversion(e)) => Err(Error::to_sql(e, error_idx)),
+        Err(frontend::BindError::Serialization(e)) => Err(Error::encode(e)),
+    }
 }
 
 pin_project! {

libs/proxy/tokio-postgres2/src/simple_query.rs

@@ -7,7 +7,6 @@ use bytes::Bytes;
 use fallible_iterator::FallibleIterator;
 use futures_util::{Stream, ready};
 use pin_project_lite::pin_project;
-use postgres_protocol2::CSafeStr;
 use postgres_protocol2::message::backend::Message;
 use postgres_protocol2::message::frontend;
 use tracing::debug;
@@ -70,9 +69,8 @@ pub async fn batch_execute(
 }
 
 pub(crate) fn encode(client: &InnerClient, query: &str) -> Result<Bytes, Error> {
-    let query = CSafeStr::new(query.as_bytes()).map_err(Error::encode)?;
     client.with_buf(|buf| {
-        frontend::query(query, buf);
+        frontend::query(query, buf).map_err(Error::encode)?;
         Ok(buf.split().freeze())
     })
 }

libs/proxy/tokio-postgres2/src/statement.rs

@@ -1,10 +1,9 @@
-use std::ffi::CStr;
 use std::fmt;
 use std::sync::{Arc, Weak};
 
+use postgres_protocol2::Oid;
 use postgres_protocol2::message::backend::Field;
 use postgres_protocol2::message::frontend;
-use postgres_protocol2::{CSafeStr, Oid};
 
 use crate::client::InnerClient;
 use crate::codec::FrontendMessage;
@@ -13,7 +12,7 @@ use crate::types::Type;
 
 struct StatementInner {
     client: Weak<InnerClient>,
-    name: &'static CStr,
+    name: &'static str,
     params: Vec<Type>,
     columns: Vec<Column>,
 }
@@ -22,7 +21,7 @@ impl Drop for StatementInner {
     fn drop(&mut self) {
         if let Some(client) = self.client.upgrade() {
             let buf = client.with_buf(|buf| {
-                frontend::close(b'S', self.name.into(), buf);
+                frontend::close(b'S', self.name, buf).unwrap();
                 frontend::sync(buf);
                 buf.split().freeze()
             });
@@ -40,7 +39,7 @@ pub struct Statement(Arc<StatementInner>);
 impl Statement {
     pub(crate) fn new(
         inner: &Arc<InnerClient>,
-        name: &'static CStr,
+        name: &'static str,
         params: Vec<Type>,
         columns: Vec<Column>,
     ) -> Statement {
@@ -55,14 +54,14 @@ impl Statement {
     pub(crate) fn new_anonymous(params: Vec<Type>, columns: Vec<Column>) -> Statement {
         Statement(Arc::new(StatementInner {
             client: Weak::new(),
-            name: c"<anonymous>",
+            name: "<anonymous>",
             params,
             columns,
         }))
     }
 
-    pub(crate) fn name(&self) -> &CSafeStr {
-        self.0.name.into()
+    pub(crate) fn name(&self) -> &str {
+        self.0.name
     }
 
     /// Returns the expected types of the statement's parameters.

libs/proxy/tokio-postgres2/src/transaction.rs

@@ -21,7 +21,7 @@ impl Drop for Transaction<'_> {
         }
 
         let buf = self.client.inner().with_buf(|buf| {
-            frontend::query(c"ROLLBACK".into(), buf);
+            frontend::query("ROLLBACK", buf).unwrap();
             buf.split().freeze()
         });
         let _ = self

pageserver/Cargo.toml

@@ -30,7 +30,6 @@ crc32c.workspace = true
 either.workspace = true
 fail.workspace = true
 futures.workspace = true
-hashlink.workspace = true
 hex.workspace = true
 humantime.workspace = true
 humantime-serde.workspace = true

pageserver/page_api/build.rs

@@ -1,13 +1,7 @@
-use std::env;
-use std::path::PathBuf;
-
-/// Generates Rust code from .proto Protobuf schemas, along with a binary file
-/// descriptor set for Protobuf schema reflection.
 fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let out_dir = PathBuf::from(env::var("OUT_DIR")?);
+    // Generates Rust code from .proto Protobuf schemas.
     tonic_build::configure()
         .bytes(["."])
-        .file_descriptor_set_path(out_dir.join("page_api_descriptor.bin"))
         .compile_protos(&["proto/page_service.proto"], &["proto"])
         .map_err(|err| err.into())
 }

pageserver/page_api/proto/page_service.proto

@@ -11,19 +11,6 @@
 // - neon-shard-id: shard ID, as <number><count> in hex ("0b10" = shard 11 of 16, 0-based)
 // - neon-timeline-id: timeline ID ("f08c4e9a2d5f76b1e3a7c2d8910f4b3e")
 //
-// The service can be accessed via e.g. grpcurl:
-//
-//    ```
-//    grpcurl \
-//      -plaintext \
-//      -H "neon-tenant-id: 7c4a1f9e3bd6470c8f3e21a65bd2e980" \
-//      -H "neon-shard-id: 0b10" \
-//      -H "neon-timeline-id: f08c4e9a2d5f76b1e3a7c2d8910f4b3e" \
-//      -H "authorization: Bearer $JWT" \
-//      -d '{"read_lsn": {"request_lsn": 1234567890}, "rel": {"spc_oid": 1663, "db_oid": 1234, "rel_number": 5678, "fork_number": 0}}'
-//      localhost:51051 page_api.PageService/CheckRelExists
-//    ```
-//
 // TODO: consider adding neon-compute-mode ("primary", "static", "replica").
 // However, this will require reconnecting when changing modes.
 //
@@ -33,7 +20,7 @@
 // - Compression
 
 syntax = "proto3";
-package page_api;
+package page_service;
 
 service PageService {
   // Returns whether a relation exists.

pageserver/page_api/src/lib.rs

@@ -7,12 +7,7 @@
 
 // Code generated by protobuf.
 pub mod proto {
-    tonic::include_proto!("page_api");
-
-    /// File descriptor set for Protobuf schema reflection. This allows using
-    /// e.g. grpcurl with the API.
-    pub const FILE_DESCRIPTOR_SET: &[u8] =
-        tonic::include_file_descriptor_set!("page_api_descriptor");
+    tonic::include_proto!("page_service");
 
     pub use page_service_client::PageServiceClient;
     pub use page_service_server::{PageService, PageServiceServer};

pageserver/src/basebackup.rs

@@ -144,7 +144,7 @@ where
         replica,
         ctx,
         io_concurrency: IoConcurrency::spawn_from_conf(
-            timeline.conf.get_vectored_concurrent_io,
+            timeline.conf,
             timeline
                 .gate
                 .enter()
@@ -343,7 +343,7 @@ where
             // Gather non-relational files from object storage pages.
             let slru_partitions = self
                 .timeline
-                .get_slru_keyspace(Version::at(self.lsn), self.ctx)
+                .get_slru_keyspace(Version::Lsn(self.lsn), self.ctx)
                 .await?
                 .partition(
                     self.timeline.get_shard_identity(),
@@ -378,7 +378,7 @@ where
             // Otherwise only include init forks of unlogged relations.
             let rels = self
                 .timeline
-                .list_rels(spcnode, dbnode, Version::at(self.lsn), self.ctx)
+                .list_rels(spcnode, dbnode, Version::Lsn(self.lsn), self.ctx)
                 .await?;
             for &rel in rels.iter() {
                 // Send init fork as main fork to provide well formed empty
@@ -517,7 +517,7 @@ where
     async fn add_rel(&mut self, src: RelTag, dst: RelTag) -> Result<(), BasebackupError> {
         let nblocks = self
             .timeline
-            .get_rel_size(src, Version::at(self.lsn), self.ctx)
+            .get_rel_size(src, Version::Lsn(self.lsn), self.ctx)
             .await?;
 
         // If the relation is empty, create an empty file
@@ -577,7 +577,7 @@ where
         let relmap_img = if has_relmap_file {
             let img = self
                 .timeline
-                .get_relmap_file(spcnode, dbnode, Version::at(self.lsn), self.ctx)
+                .get_relmap_file(spcnode, dbnode, Version::Lsn(self.lsn), self.ctx)
                 .await?;
 
             if img.len()
@@ -631,7 +631,7 @@ where
             if !has_relmap_file
                 && self
                     .timeline
-                    .list_rels(spcnode, dbnode, Version::at(self.lsn), self.ctx)
+                    .list_rels(spcnode, dbnode, Version::Lsn(self.lsn), self.ctx)
                     .await?
                     .is_empty()
             {

pageserver/src/http/routes.rs

@@ -3199,7 +3199,7 @@ async fn list_aux_files(
             .await?;
 
     let io_concurrency = IoConcurrency::spawn_from_conf(
-        state.conf.get_vectored_concurrent_io,
+        state.conf,
         timeline.gate.enter().map_err(|_| ApiError::Cancelled)?,
     );
 

pageserver/src/metrics.rs

@@ -843,50 +843,23 @@ pub(crate) static COMPRESSION_IMAGE_OUTPUT_BYTES: Lazy<IntCounter> = Lazy::new(|
     .expect("failed to define a metric")
 });
 
-pub(crate) static RELSIZE_LATEST_CACHE_ENTRIES: Lazy<UIntGauge> = Lazy::new(|| {
+pub(crate) static RELSIZE_CACHE_ENTRIES: Lazy<UIntGauge> = Lazy::new(|| {
     register_uint_gauge!(
-        "pageserver_relsize_latest_cache_entries",
-        "Number of entries in the latest relation size cache",
+        "pageserver_relsize_cache_entries",
+        "Number of entries in the relation size cache",
     )
     .expect("failed to define a metric")
 });
 
-pub(crate) static RELSIZE_LATEST_CACHE_HITS: Lazy<IntCounter> = Lazy::new(|| {
+pub(crate) static RELSIZE_CACHE_HITS: Lazy<IntCounter> = Lazy::new(|| {
+    register_int_counter!("pageserver_relsize_cache_hits", "Relation size cache hits",)
+        .expect("failed to define a metric")
+});
+
+pub(crate) static RELSIZE_CACHE_MISSES: Lazy<IntCounter> = Lazy::new(|| {
     register_int_counter!(
-        "pageserver_relsize_latest_cache_hits",
-        "Latest relation size cache hits",
-    )
-    .expect("failed to define a metric")
-});
-
-pub(crate) static RELSIZE_LATEST_CACHE_MISSES: Lazy<IntCounter> = Lazy::new(|| {
-    register_int_counter!(
-        "pageserver_relsize_latest_cache_misses",
-        "Relation size latest cache misses",
-    )
-    .expect("failed to define a metric")
-});
-
-pub(crate) static RELSIZE_SNAPSHOT_CACHE_ENTRIES: Lazy<UIntGauge> = Lazy::new(|| {
-    register_uint_gauge!(
-        "pageserver_relsize_snapshot_cache_entries",
-        "Number of entries in the pitr relation size cache",
-    )
-    .expect("failed to define a metric")
-});
-
-pub(crate) static RELSIZE_SNAPSHOT_CACHE_HITS: Lazy<IntCounter> = Lazy::new(|| {
-    register_int_counter!(
-        "pageserver_relsize_snapshot_cache_hits",
-        "Pitr relation size cache hits",
-    )
-    .expect("failed to define a metric")
-});
-
-pub(crate) static RELSIZE_SNAPSHOT_CACHE_MISSES: Lazy<IntCounter> = Lazy::new(|| {
-    register_int_counter!(
-        "pageserver_relsize_snapshot_cache_misses",
-        "Relation size snapshot cache misses",
+        "pageserver_relsize_cache_misses",
+        "Relation size cache misses",
     )
     .expect("failed to define a metric")
 });

pageserver/src/page_service.rs

@@ -18,7 +18,7 @@ use itertools::Itertools;
 use jsonwebtoken::TokenData;
 use once_cell::sync::OnceCell;
 use pageserver_api::config::{
-    GetVectoredConcurrentIo, PageServicePipeliningConfig, PageServicePipeliningConfigPipelined,
+    PageServicePipeliningConfig, PageServicePipeliningConfigPipelined,
     PageServiceProtocolPipelinedBatchingStrategy, PageServiceProtocolPipelinedExecutionStrategy,
 };
 use pageserver_api::key::rel_block_to_key;
@@ -62,7 +62,7 @@ use crate::metrics::{
     self, COMPUTE_COMMANDS_COUNTERS, ComputeCommandKind, GetPageBatchBreakReason, LIVE_CONNECTIONS,
     SmgrOpTimer, TimelineMetrics,
 };
-use crate::pgdatadir_mapping::{LsnRange, Version};
+use crate::pgdatadir_mapping::Version;
 use crate::span::{
     debug_assert_current_span_has_tenant_and_timeline_id,
     debug_assert_current_span_has_tenant_and_timeline_id_no_shard_id,
@@ -331,10 +331,10 @@ async fn page_service_conn_main(
     // But it's in a shared crate, so, we store connection_ctx inside PageServerHandler
     // and create the per-query context in process_query ourselves.
     let mut conn_handler = PageServerHandler::new(
+        conf,
         tenant_manager,
         auth,
         pipelining_config,
-        conf.get_vectored_concurrent_io,
         perf_span_fields,
         connection_ctx,
         cancel.clone(),
@@ -371,6 +371,7 @@ async fn page_service_conn_main(
 }
 
 struct PageServerHandler {
+    conf: &'static PageServerConf,
     auth: Option<Arc<SwappableJwtAuth>>,
     claims: Option<Claims>,
 
@@ -388,7 +389,6 @@ struct PageServerHandler {
     timeline_handles: Option<TimelineHandles>,
 
     pipelining_config: PageServicePipeliningConfig,
-    get_vectored_concurrent_io: GetVectoredConcurrentIo,
 
     gate_guard: GateGuard,
 }
@@ -642,7 +642,7 @@ impl std::fmt::Display for BatchedPageStreamError {
 struct BatchedGetPageRequest {
     req: PagestreamGetPageRequest,
     timer: SmgrOpTimer,
-    lsn_range: LsnRange,
+    effective_request_lsn: Lsn,
     ctx: RequestContext,
 }
 
@@ -764,12 +764,12 @@ impl BatchedFeMessage {
                 match batching_strategy {
                     PageServiceProtocolPipelinedBatchingStrategy::UniformLsn => {
                         if let Some(last_in_batch) = accum_pages.last() {
-                            if last_in_batch.lsn_range.effective_lsn
-                                != this_pages[0].lsn_range.effective_lsn
+                            if last_in_batch.effective_request_lsn
+                                != this_pages[0].effective_request_lsn
                             {
                                 trace!(
-                                    accum_lsn = %last_in_batch.lsn_range.effective_lsn,
-                                    this_lsn = %this_pages[0].lsn_range.effective_lsn,
+                                    accum_lsn = %last_in_batch.effective_request_lsn,
+                                    this_lsn = %this_pages[0].effective_request_lsn,
                                     "stopping batching because LSN changed"
                                 );
 
@@ -784,15 +784,15 @@ impl BatchedFeMessage {
                         let same_page_different_lsn = accum_pages.iter().any(|batched| {
                             batched.req.rel == this_pages[0].req.rel
                                 && batched.req.blkno == this_pages[0].req.blkno
-                                && batched.lsn_range.effective_lsn
-                                    != this_pages[0].lsn_range.effective_lsn
+                                && batched.effective_request_lsn
+                                    != this_pages[0].effective_request_lsn
                         });
 
                         if same_page_different_lsn {
                             trace!(
                                 rel=%this_pages[0].req.rel,
                                 blkno=%this_pages[0].req.blkno,
-                                lsn=%this_pages[0].lsn_range.effective_lsn,
+                                lsn=%this_pages[0].effective_request_lsn,
                                 "stopping batching because same page was requested at different LSNs"
                             );
 
@@ -844,16 +844,17 @@ impl BatchedFeMessage {
 impl PageServerHandler {
     #[allow(clippy::too_many_arguments)]
     pub fn new(
+        conf: &'static PageServerConf,
         tenant_manager: Arc<TenantManager>,
         auth: Option<Arc<SwappableJwtAuth>>,
         pipelining_config: PageServicePipeliningConfig,
-        get_vectored_concurrent_io: GetVectoredConcurrentIo,
         perf_span_fields: ConnectionPerfSpanFields,
         connection_ctx: RequestContext,
         cancel: CancellationToken,
         gate_guard: GateGuard,
     ) -> Self {
         PageServerHandler {
+            conf,
             auth,
             claims: None,
             connection_ctx,
@@ -861,7 +862,6 @@ impl PageServerHandler {
             timeline_handles: Some(TimelineHandles::new(tenant_manager)),
             cancel,
             pipelining_config,
-            get_vectored_concurrent_io,
             gate_guard,
         }
     }
@@ -1158,7 +1158,7 @@ impl PageServerHandler {
                 .await?;
 
                 // We're holding the Handle
-                let effective_lsn = match Self::effective_request_lsn(
+                let effective_request_lsn = match Self::effective_request_lsn(
                     &shard,
                     shard.get_last_record_lsn(),
                     req.hdr.request_lsn,
@@ -1177,10 +1177,7 @@ impl PageServerHandler {
                     pages: smallvec::smallvec![BatchedGetPageRequest {
                         req,
                         timer,
-                        lsn_range: LsnRange {
-                            effective_lsn,
-                            request_lsn: req.hdr.request_lsn
-                        },
+                        effective_request_lsn,
                         ctx,
                     }],
                     // The executor grabs the batch when it becomes idle.
@@ -1626,7 +1623,7 @@ impl PageServerHandler {
         }
 
         let io_concurrency = IoConcurrency::spawn_from_conf(
-            self.get_vectored_concurrent_io,
+            self.conf,
             match self.gate_guard.try_clone() {
                 Ok(guard) => guard,
                 Err(_) => {
@@ -2130,14 +2127,7 @@ impl PageServerHandler {
         .await?;
 
         let exists = timeline
-            .get_rel_exists(
-                req.rel,
-                Version::LsnRange(LsnRange {
-                    effective_lsn: lsn,
-                    request_lsn: req.hdr.request_lsn,
-                }),
-                ctx,
-            )
+            .get_rel_exists(req.rel, Version::Lsn(lsn), ctx)
             .await?;
 
         Ok(PagestreamBeMessage::Exists(PagestreamExistsResponse {
@@ -2164,14 +2154,7 @@ impl PageServerHandler {
         .await?;
 
         let n_blocks = timeline
-            .get_rel_size(
-                req.rel,
-                Version::LsnRange(LsnRange {
-                    effective_lsn: lsn,
-                    request_lsn: req.hdr.request_lsn,
-                }),
-                ctx,
-            )
+            .get_rel_size(req.rel, Version::Lsn(lsn), ctx)
             .await?;
 
         Ok(PagestreamBeMessage::Nblocks(PagestreamNblocksResponse {
@@ -2198,15 +2181,7 @@ impl PageServerHandler {
         .await?;
 
         let total_blocks = timeline
-            .get_db_size(
-                DEFAULTTABLESPACE_OID,
-                req.dbnode,
-                Version::LsnRange(LsnRange {
-                    effective_lsn: lsn,
-                    request_lsn: req.hdr.request_lsn,
-                }),
-                ctx,
-            )
+            .get_db_size(DEFAULTTABLESPACE_OID, req.dbnode, Version::Lsn(lsn), ctx)
             .await?;
         let db_size = total_blocks as i64 * BLCKSZ as i64;
 
@@ -2239,7 +2214,7 @@ impl PageServerHandler {
                 // Ignore error (trace buffer may be full or tracer may have disconnected).
                 _ = page_trace.try_send(PageTraceEvent {
                     key,
-                    effective_lsn: batch.lsn_range.effective_lsn,
+                    effective_lsn: batch.effective_request_lsn,
                     time,
                 });
             }
@@ -2254,7 +2229,7 @@ impl PageServerHandler {
                     perf_instrument = true;
                 }
 
-                req.lsn_range.effective_lsn
+                req.effective_request_lsn
             })
             .max()
             .expect("batch is never empty");
@@ -2308,7 +2283,7 @@ impl PageServerHandler {
                     (
                         &p.req.rel,
                         &p.req.blkno,
-                        p.lsn_range,
+                        p.effective_request_lsn,
                         p.ctx.attached_child(),
                     )
                 }),

pageserver/src/pgdatadir_mapping.rs

@@ -43,9 +43,7 @@ use crate::aux_file;
 use crate::context::{PerfInstrumentFutureExt, RequestContext, RequestContextBuilder};
 use crate::keyspace::{KeySpace, KeySpaceAccum};
 use crate::metrics::{
-    RELSIZE_CACHE_MISSES_OLD, RELSIZE_LATEST_CACHE_ENTRIES, RELSIZE_LATEST_CACHE_HITS,
-    RELSIZE_LATEST_CACHE_MISSES, RELSIZE_SNAPSHOT_CACHE_ENTRIES, RELSIZE_SNAPSHOT_CACHE_HITS,
-    RELSIZE_SNAPSHOT_CACHE_MISSES,
+    RELSIZE_CACHE_ENTRIES, RELSIZE_CACHE_HITS, RELSIZE_CACHE_MISSES, RELSIZE_CACHE_MISSES_OLD,
 };
 use crate::span::{
     debug_assert_current_span_has_tenant_and_timeline_id,
@@ -92,28 +90,6 @@ pub enum LsnForTimestamp {
     NoData(Lsn),
 }
 
-/// Each request to page server contains LSN range: `not_modified_since..request_lsn`.
-/// See comments libs/pageserver_api/src/models.rs.
-/// Based on this range and `last_record_lsn` PS calculates `effective_lsn`.
-/// But to distinguish requests from primary and replicas we need also to pass `request_lsn`.
-#[derive(Debug, Clone, Copy, Default)]
-pub struct LsnRange {
-    pub effective_lsn: Lsn,
-    pub request_lsn: Lsn,
-}
-
-impl LsnRange {
-    pub fn at(lsn: Lsn) -> LsnRange {
-        LsnRange {
-            effective_lsn: lsn,
-            request_lsn: lsn,
-        }
-    }
-    pub fn is_latest(&self) -> bool {
-        self.request_lsn == Lsn::MAX
-    }
-}
-
 #[derive(Debug, thiserror::Error)]
 pub(crate) enum CalculateLogicalSizeError {
     #[error("cancelled")]
@@ -226,13 +202,13 @@ impl Timeline {
         io_concurrency: IoConcurrency,
     ) -> Result<Bytes, PageReconstructError> {
         match version {
-            Version::LsnRange(lsns) => {
+            Version::Lsn(effective_lsn) => {
                 let pages: smallvec::SmallVec<[_; 1]> = smallvec::smallvec![(tag, blknum)];
                 let res = self
                     .get_rel_page_at_lsn_batched(
-                        pages
-                            .iter()
-                            .map(|(tag, blknum)| (tag, blknum, lsns, ctx.attached_child())),
+                        pages.iter().map(|(tag, blknum)| {
+                            (tag, blknum, effective_lsn, ctx.attached_child())
+                        }),
                         io_concurrency.clone(),
                         ctx,
                     )
@@ -270,7 +246,7 @@ impl Timeline {
     /// The ordering of the returned vec corresponds to the ordering of `pages`.
     pub(crate) async fn get_rel_page_at_lsn_batched(
         &self,
-        pages: impl ExactSizeIterator<Item = (&RelTag, &BlockNumber, LsnRange, RequestContext)>,
+        pages: impl ExactSizeIterator<Item = (&RelTag, &BlockNumber, Lsn, RequestContext)>,
         io_concurrency: IoConcurrency,
         ctx: &RequestContext,
     ) -> Vec<Result<Bytes, PageReconstructError>> {
@@ -289,7 +265,7 @@ impl Timeline {
         let mut req_keyspaces: HashMap<Lsn, KeySpaceRandomAccum> =
             HashMap::with_capacity(pages.len());
 
-        for (response_slot_idx, (tag, blknum, lsns, ctx)) in pages.enumerate() {
+        for (response_slot_idx, (tag, blknum, lsn, ctx)) in pages.enumerate() {
             if tag.relnode == 0 {
                 result_slots[response_slot_idx].write(Err(PageReconstructError::Other(
                     RelationError::InvalidRelnode.into(),
@@ -298,7 +274,7 @@ impl Timeline {
                 slots_filled += 1;
                 continue;
             }
-            let lsn = lsns.effective_lsn;
+
             let nblocks = {
                 let ctx = RequestContextBuilder::from(&ctx)
                     .perf_span(|crnt_perf_span| {
@@ -313,7 +289,7 @@ impl Timeline {
                     .attached_child();
 
                 match self
-                    .get_rel_size(*tag, Version::LsnRange(lsns), &ctx)
+                    .get_rel_size(*tag, Version::Lsn(lsn), &ctx)
                     .maybe_perf_instrument(&ctx, |crnt_perf_span| crnt_perf_span.clone())
                     .await
                 {
@@ -494,7 +470,7 @@ impl Timeline {
             ));
         }
 
-        if let Some(nblocks) = self.get_cached_rel_size(&tag, version) {
+        if let Some(nblocks) = self.get_cached_rel_size(&tag, version.get_lsn()) {
             return Ok(nblocks);
         }
 
@@ -512,7 +488,7 @@ impl Timeline {
         let mut buf = version.get(self, key, ctx).await?;
         let nblocks = buf.get_u32_le();
 
-        self.update_cached_rel_size(tag, version, nblocks);
+        self.update_cached_rel_size(tag, version.get_lsn(), nblocks);
 
         Ok(nblocks)
     }
@@ -534,7 +510,7 @@ impl Timeline {
         }
 
         // first try to lookup relation in cache
-        if let Some(_nblocks) = self.get_cached_rel_size(&tag, version) {
+        if let Some(_nblocks) = self.get_cached_rel_size(&tag, version.get_lsn()) {
             return Ok(true);
         }
         // then check if the database was already initialized.
@@ -610,7 +586,7 @@ impl Timeline {
         // scan directory listing (new), merge with the old results
         let key_range = rel_tag_sparse_key_range(spcnode, dbnode);
         let io_concurrency = IoConcurrency::spawn_from_conf(
-            self.conf.get_vectored_concurrent_io,
+            self.conf,
             self.gate
                 .enter()
                 .map_err(|_| PageReconstructError::Cancelled)?,
@@ -656,7 +632,7 @@ impl Timeline {
     ) -> Result<Bytes, PageReconstructError> {
         assert!(self.tenant_shard_id.is_shard_zero());
         let n_blocks = self
-            .get_slru_segment_size(kind, segno, Version::at(lsn), ctx)
+            .get_slru_segment_size(kind, segno, Version::Lsn(lsn), ctx)
             .await?;
 
         let keyspace = KeySpace::single(
@@ -669,7 +645,7 @@ impl Timeline {
         );
 
         let io_concurrency = IoConcurrency::spawn_from_conf(
-            self.conf.get_vectored_concurrent_io,
+            self.conf,
             self.gate
                 .enter()
                 .map_err(|_| PageReconstructError::Cancelled)?,
@@ -891,11 +867,11 @@ impl Timeline {
         mut f: impl FnMut(TimestampTz) -> ControlFlow<T>,
     ) -> Result<T, PageReconstructError> {
         for segno in self
-            .list_slru_segments(SlruKind::Clog, Version::at(probe_lsn), ctx)
+            .list_slru_segments(SlruKind::Clog, Version::Lsn(probe_lsn), ctx)
             .await?
         {
             let nblocks = self
-                .get_slru_segment_size(SlruKind::Clog, segno, Version::at(probe_lsn), ctx)
+                .get_slru_segment_size(SlruKind::Clog, segno, Version::Lsn(probe_lsn), ctx)
                 .await?;
 
             let keyspace = KeySpace::single(
@@ -909,7 +885,7 @@ impl Timeline {
             );
 
             let io_concurrency = IoConcurrency::spawn_from_conf(
-                self.conf.get_vectored_concurrent_io,
+                self.conf,
                 self.gate
                     .enter()
                     .map_err(|_| PageReconstructError::Cancelled)?,
@@ -1161,7 +1137,7 @@ impl Timeline {
         let mut total_size: u64 = 0;
         for (spcnode, dbnode) in dbdir.dbdirs.keys() {
             for rel in self
-                .list_rels(*spcnode, *dbnode, Version::at(lsn), ctx)
+                .list_rels(*spcnode, *dbnode, Version::Lsn(lsn), ctx)
                 .await?
             {
                 if self.cancel.is_cancelled() {
@@ -1236,7 +1212,7 @@ impl Timeline {
             result.add_key(rel_dir_to_key(spcnode, dbnode));
 
             let mut rels: Vec<RelTag> = self
-                .list_rels(spcnode, dbnode, Version::at(lsn), ctx)
+                .list_rels(spcnode, dbnode, Version::Lsn(lsn), ctx)
                 .await?
                 .into_iter()
                 .collect();
@@ -1353,75 +1329,59 @@ impl Timeline {
         Ok((dense_keyspace, sparse_keyspace))
     }
 
-    /// Get cached size of relation. There are two caches: one for primary updates, it captures the latest state of
-    /// of the timeline and snapshot cache, which key includes LSN and so can be used by replicas to get relation size
-    /// at the particular LSN (snapshot).
-    pub fn get_cached_rel_size(&self, tag: &RelTag, version: Version<'_>) -> Option<BlockNumber> {
-        let lsn = version.get_lsn();
-        {
-            let rel_size_cache = self.rel_size_latest_cache.read().unwrap();
-            if let Some((cached_lsn, nblocks)) = rel_size_cache.get(tag) {
-                if lsn >= *cached_lsn {
-                    RELSIZE_LATEST_CACHE_HITS.inc();
-                    return Some(*nblocks);
-                }
-                RELSIZE_CACHE_MISSES_OLD.inc();
+    /// Get cached size of relation if it not updated after specified LSN
+    pub fn get_cached_rel_size(&self, tag: &RelTag, lsn: Lsn) -> Option<BlockNumber> {
+        let rel_size_cache = self.rel_size_cache.read().unwrap();
+        if let Some((cached_lsn, nblocks)) = rel_size_cache.map.get(tag) {
+            if lsn >= *cached_lsn {
+                RELSIZE_CACHE_HITS.inc();
+                return Some(*nblocks);
             }
+            RELSIZE_CACHE_MISSES_OLD.inc();
         }
-        {
-            let mut rel_size_cache = self.rel_size_snapshot_cache.lock().unwrap();
-            if let Some(nblock) = rel_size_cache.get(&(lsn, *tag)) {
-                RELSIZE_SNAPSHOT_CACHE_HITS.inc();
-                return Some(*nblock);
-            }
-        }
-        if version.is_latest() {
-            RELSIZE_LATEST_CACHE_MISSES.inc();
-        } else {
-            RELSIZE_SNAPSHOT_CACHE_MISSES.inc();
-        }
+        RELSIZE_CACHE_MISSES.inc();
         None
     }
 
     /// Update cached relation size if there is no more recent update
-    pub fn update_cached_rel_size(&self, tag: RelTag, version: Version<'_>, nblocks: BlockNumber) {
-        let lsn = version.get_lsn();
-        if version.is_latest() {
-            let mut rel_size_cache = self.rel_size_latest_cache.write().unwrap();
-            match rel_size_cache.entry(tag) {
-                hash_map::Entry::Occupied(mut entry) => {
-                    let cached_lsn = entry.get_mut();
-                    if lsn >= cached_lsn.0 {
-                        *cached_lsn = (lsn, nblocks);
-                    }
-                }
-                hash_map::Entry::Vacant(entry) => {
-                    entry.insert((lsn, nblocks));
-                    RELSIZE_LATEST_CACHE_ENTRIES.inc();
+    pub fn update_cached_rel_size(&self, tag: RelTag, lsn: Lsn, nblocks: BlockNumber) {
+        let mut rel_size_cache = self.rel_size_cache.write().unwrap();
+
+        if lsn < rel_size_cache.complete_as_of {
+            // Do not cache old values. It's safe to cache the size on read, as long as
+            // the read was at an LSN since we started the WAL ingestion. Reasoning: we
+            // never evict values from the cache, so if the relation size changed after
+            // 'lsn', the new value is already in the cache.
+            return;
+        }
+
+        match rel_size_cache.map.entry(tag) {
+            hash_map::Entry::Occupied(mut entry) => {
+                let cached_lsn = entry.get_mut();
+                if lsn >= cached_lsn.0 {
+                    *cached_lsn = (lsn, nblocks);
                 }
             }
-        } else {
-            let mut rel_size_cache = self.rel_size_snapshot_cache.lock().unwrap();
-            if rel_size_cache.capacity() != 0 {
-                rel_size_cache.insert((lsn, tag), nblocks);
-                RELSIZE_SNAPSHOT_CACHE_ENTRIES.set(rel_size_cache.len() as u64);
+            hash_map::Entry::Vacant(entry) => {
+                entry.insert((lsn, nblocks));
+                RELSIZE_CACHE_ENTRIES.inc();
             }
         }
     }
 
     /// Store cached relation size
     pub fn set_cached_rel_size(&self, tag: RelTag, lsn: Lsn, nblocks: BlockNumber) {
-        let mut rel_size_cache = self.rel_size_latest_cache.write().unwrap();
-        if rel_size_cache.insert(tag, (lsn, nblocks)).is_none() {
-            RELSIZE_LATEST_CACHE_ENTRIES.inc();
+        let mut rel_size_cache = self.rel_size_cache.write().unwrap();
+        if rel_size_cache.map.insert(tag, (lsn, nblocks)).is_none() {
+            RELSIZE_CACHE_ENTRIES.inc();
         }
     }
 
     /// Remove cached relation size
     pub fn remove_cached_rel_size(&self, tag: &RelTag) {
-        let mut rel_size_cache = self.rel_size_latest_cache.write().unwrap();
-        if rel_size_cache.remove(tag).is_some() {
-            RELSIZE_LATEST_CACHE_ENTRIES.dec();
+        let mut rel_size_cache = self.rel_size_cache.write().unwrap();
+        if rel_size_cache.map.remove(tag).is_some() {
+            RELSIZE_CACHE_ENTRIES.dec();
         }
     }
 }
@@ -1625,10 +1585,7 @@ impl DatadirModification<'_> {
         //       check the cache too. This is because eagerly checking the cache results in
         //       less work overall and 10% better performance. It's more work on cache miss
         //       but cache miss is rare.
-        if let Some(nblocks) = self
-            .tline
-            .get_cached_rel_size(&rel, Version::Modified(self))
-        {
+        if let Some(nblocks) = self.tline.get_cached_rel_size(&rel, self.get_lsn()) {
             Ok(nblocks)
         } else if !self
             .tline
@@ -2710,7 +2667,7 @@ pub struct DatadirModificationStats {
 /// timeline to not miss the latest updates.
 #[derive(Clone, Copy)]
 pub enum Version<'a> {
-    LsnRange(LsnRange),
+    Lsn(Lsn),
     Modified(&'a DatadirModification<'a>),
 }
 
@@ -2722,7 +2679,7 @@ impl Version<'_> {
         ctx: &RequestContext,
     ) -> Result<Bytes, PageReconstructError> {
         match self {
-            Version::LsnRange(lsns) => timeline.get(key, lsns.effective_lsn, ctx).await,
+            Version::Lsn(lsn) => timeline.get(key, *lsn, ctx).await,
             Version::Modified(modification) => modification.get(key, ctx).await,
         }
     }
@@ -2744,26 +2701,12 @@ impl Version<'_> {
         }
     }
 
-    pub fn is_latest(&self) -> bool {
+    fn get_lsn(&self) -> Lsn {
         match self {
-            Version::LsnRange(lsns) => lsns.is_latest(),
-            Version::Modified(_) => true,
-        }
-    }
-
-    pub fn get_lsn(&self) -> Lsn {
-        match self {
-            Version::LsnRange(lsns) => lsns.effective_lsn,
+            Version::Lsn(lsn) => *lsn,
             Version::Modified(modification) => modification.lsn,
         }
     }
-
-    pub fn at(lsn: Lsn) -> Self {
-        Version::LsnRange(LsnRange {
-            effective_lsn: lsn,
-            request_lsn: lsn,
-        })
-    }
 }
 
 //--- Metadata structs stored in key-value pairs in the repository.

pageserver/src/tenant.rs

@@ -8596,10 +8596,8 @@ mod tests {
         lsn: Lsn,
         ctx: &RequestContext,
     ) -> Result<Option<Bytes>, GetVectoredError> {
-        let io_concurrency = IoConcurrency::spawn_from_conf(
-            tline.conf.get_vectored_concurrent_io,
-            tline.gate.enter().unwrap(),
-        );
+        let io_concurrency =
+            IoConcurrency::spawn_from_conf(tline.conf, tline.gate.enter().unwrap());
         let mut reconstruct_state = ValuesReconstructState::new(io_concurrency);
         let query = VersionedKeySpaceQuery::uniform(KeySpace::single(key..key.next()), lsn);
         let mut res = tline

pageserver/src/tenant/storage_layer.rs

@@ -31,7 +31,6 @@ pub use inmemory_layer::InMemoryLayer;
 pub(crate) use layer::{EvictionError, Layer, ResidentLayer};
 pub use layer_desc::{PersistentLayerDesc, PersistentLayerKey};
 pub use layer_name::{DeltaLayerName, ImageLayerName, LayerName};
-use pageserver_api::config::GetVectoredConcurrentIo;
 use pageserver_api::key::Key;
 use pageserver_api::keyspace::{KeySpace, KeySpaceRandomAccum};
 use pageserver_api::record::NeonWalRecord;
@@ -44,6 +43,7 @@ use self::inmemory_layer::InMemoryLayerFileId;
 use super::PageReconstructError;
 use super::layer_map::InMemoryLayerDesc;
 use super::timeline::{GetVectoredError, ReadPath};
+use crate::config::PageServerConf;
 use crate::context::{
     AccessStatsBehavior, PerfInstrumentFutureExt, RequestContext, RequestContextBuilder,
 };
@@ -318,10 +318,11 @@ impl IoConcurrency {
     }
 
     pub(crate) fn spawn_from_conf(
-        conf: GetVectoredConcurrentIo,
+        conf: &'static PageServerConf,
         gate_guard: GateGuard,
     ) -> IoConcurrency {
-        let selected = match conf {
+        use pageserver_api::config::GetVectoredConcurrentIo;
+        let selected = match conf.get_vectored_concurrent_io {
             GetVectoredConcurrentIo::Sequential => SelectedIoConcurrency::Sequential,
             GetVectoredConcurrentIo::SidecarTask => SelectedIoConcurrency::SidecarTask(gate_guard),
         };

pageserver/src/tenant/timeline.rs

@@ -14,7 +14,6 @@ pub mod span;
 pub mod uninit;
 mod walreceiver;
 
-use hashlink::LruCache;
 use std::array;
 use std::cmp::{max, min};
 use std::collections::btree_map::Entry;
@@ -198,6 +197,16 @@ pub struct TimelineResources {
     pub l0_flush_global_state: l0_flush::L0FlushGlobalState,
 }
 
+/// The relation size cache caches relation sizes at the end of the timeline. It speeds up WAL
+/// ingestion considerably, because WAL ingestion needs to check on most records if the record
+/// implicitly extends the relation.  At startup, `complete_as_of` is initialized to the current end
+/// of the timeline (disk_consistent_lsn).  It's used on reads of relation sizes to check if the
+/// value can be used to also update the cache, see [`Timeline::update_cached_rel_size`].
+pub(crate) struct RelSizeCache {
+    pub(crate) complete_as_of: Lsn,
+    pub(crate) map: HashMap<RelTag, (Lsn, BlockNumber)>,
+}
+
 pub struct Timeline {
     pub(crate) conf: &'static PageServerConf,
     tenant_conf: Arc<ArcSwap<AttachedTenantConf>>,
@@ -356,8 +365,7 @@ pub struct Timeline {
     pub walreceiver: Mutex<Option<WalReceiver>>,
 
     /// Relation size cache
-    pub(crate) rel_size_latest_cache: RwLock<HashMap<RelTag, (Lsn, BlockNumber)>>,
-    pub(crate) rel_size_snapshot_cache: Mutex<LruCache<(Lsn, RelTag), BlockNumber>>,
+    pub(crate) rel_size_cache: RwLock<RelSizeCache>,
 
     download_all_remote_layers_task_info: RwLock<Option<DownloadRemoteLayersTaskInfo>>,
 
@@ -2812,13 +2820,6 @@ impl Timeline {
 
             self.remote_client.update_config(&new_conf.location);
 
-            let mut rel_size_cache = self.rel_size_snapshot_cache.lock().unwrap();
-            if let Some(new_capacity) = new_conf.tenant_conf.relsize_snapshot_cache_capacity {
-                if new_capacity != rel_size_cache.capacity() {
-                    rel_size_cache.set_capacity(new_capacity);
-                }
-            }
-
             self.metrics
                 .evictions_with_low_residence_duration
                 .write()
@@ -2877,14 +2878,6 @@ impl Timeline {
             ancestor_gc_info.insert_child(timeline_id, metadata.ancestor_lsn(), is_offloaded);
         }
 
-        let relsize_snapshot_cache_capacity = {
-            let loaded_tenant_conf = tenant_conf.load();
-            loaded_tenant_conf
-                .tenant_conf
-                .relsize_snapshot_cache_capacity
-                .unwrap_or(conf.default_tenant_conf.relsize_snapshot_cache_capacity)
-        };
-
         Arc::new_cyclic(|myself| {
             let metrics = Arc::new(TimelineMetrics::new(
                 &tenant_shard_id,
@@ -2976,8 +2969,10 @@ impl Timeline {
                 last_image_layer_creation_check_instant: Mutex::new(None),
 
                 last_received_wal: Mutex::new(None),
-                rel_size_latest_cache: RwLock::new(HashMap::new()),
-                rel_size_snapshot_cache: Mutex::new(LruCache::new(relsize_snapshot_cache_capacity)),
+                rel_size_cache: RwLock::new(RelSizeCache {
+                    complete_as_of: disk_consistent_lsn,
+                    map: HashMap::new(),
+                }),
 
                 download_all_remote_layers_task_info: RwLock::new(None),
 
@@ -3535,7 +3530,7 @@ impl Timeline {
                 };
 
                 let io_concurrency = IoConcurrency::spawn_from_conf(
-                    self_ref.conf.get_vectored_concurrent_io,
+                    self_ref.conf,
                     self_ref
                         .gate
                         .enter()
@@ -5564,7 +5559,7 @@ impl Timeline {
             });
 
             let io_concurrency = IoConcurrency::spawn_from_conf(
-                self.conf.get_vectored_concurrent_io,
+                self.conf,
                 self.gate
                     .enter()
                     .map_err(|_| CreateImageLayersError::Cancelled)?,

pageserver/src/tenant/timeline/detach_ancestor.rs

@@ -188,7 +188,7 @@ pub(crate) async fn generate_tombstone_image_layer(
         "removing non-inherited keys by writing an image layer with tombstones at the detach LSN"
     );
     let io_concurrency = IoConcurrency::spawn_from_conf(
-        detached.conf.get_vectored_concurrent_io,
+        detached.conf,
         detached.gate.enter().map_err(|_| Error::ShuttingDown)?,
     );
     let mut reconstruct_state = ValuesReconstructState::new(io_concurrency);

pageserver/src/walingest.rs

@@ -1684,31 +1684,31 @@ mod tests {
         // The relation was created at LSN 2, not visible at LSN 1 yet.
         assert_eq!(
             tline
-                .get_rel_exists(TESTREL_A, Version::at(Lsn(0x10)), &ctx)
+                .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x10)), &ctx)
                 .await?,
             false
         );
         assert!(
             tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x10)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x10)), &ctx)
                 .await
                 .is_err()
         );
         assert_eq!(
             tline
-                .get_rel_exists(TESTREL_A, Version::at(Lsn(0x20)), &ctx)
+                .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x20)), &ctx)
                 .await?,
             true
         );
         assert_eq!(
             tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x20)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x20)), &ctx)
                 .await?,
             1
         );
         assert_eq!(
             tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x50)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x50)), &ctx)
                 .await?,
             3
         );
@@ -1719,7 +1719,7 @@ mod tests {
                 .get_rel_page_at_lsn(
                     TESTREL_A,
                     0,
-                    Version::at(Lsn(0x20)),
+                    Version::Lsn(Lsn(0x20)),
                     &ctx,
                     io_concurrency.clone()
                 )
@@ -1733,7 +1733,7 @@ mod tests {
                 .get_rel_page_at_lsn(
                     TESTREL_A,
                     0,
-                    Version::at(Lsn(0x30)),
+                    Version::Lsn(Lsn(0x30)),
                     &ctx,
                     io_concurrency.clone()
                 )
@@ -1747,7 +1747,7 @@ mod tests {
                 .get_rel_page_at_lsn(
                     TESTREL_A,
                     0,
-                    Version::at(Lsn(0x40)),
+                    Version::Lsn(Lsn(0x40)),
                     &ctx,
                     io_concurrency.clone()
                 )
@@ -1760,7 +1760,7 @@ mod tests {
                 .get_rel_page_at_lsn(
                     TESTREL_A,
                     1,
-                    Version::at(Lsn(0x40)),
+                    Version::Lsn(Lsn(0x40)),
                     &ctx,
                     io_concurrency.clone()
                 )
@@ -1774,7 +1774,7 @@ mod tests {
                 .get_rel_page_at_lsn(
                     TESTREL_A,
                     0,
-                    Version::at(Lsn(0x50)),
+                    Version::Lsn(Lsn(0x50)),
                     &ctx,
                     io_concurrency.clone()
                 )
@@ -1787,7 +1787,7 @@ mod tests {
                 .get_rel_page_at_lsn(
                     TESTREL_A,
                     1,
-                    Version::at(Lsn(0x50)),
+                    Version::Lsn(Lsn(0x50)),
                     &ctx,
                     io_concurrency.clone()
                 )
@@ -1800,7 +1800,7 @@ mod tests {
                 .get_rel_page_at_lsn(
                     TESTREL_A,
                     2,
-                    Version::at(Lsn(0x50)),
+                    Version::Lsn(Lsn(0x50)),
                     &ctx,
                     io_concurrency.clone()
                 )
@@ -1820,7 +1820,7 @@ mod tests {
         // Check reported size and contents after truncation
         assert_eq!(
             tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x60)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x60)), &ctx)
                 .await?,
             2
         );
@@ -1829,7 +1829,7 @@ mod tests {
                 .get_rel_page_at_lsn(
                     TESTREL_A,
                     0,
-                    Version::at(Lsn(0x60)),
+                    Version::Lsn(Lsn(0x60)),
                     &ctx,
                     io_concurrency.clone()
                 )
@@ -1842,7 +1842,7 @@ mod tests {
                 .get_rel_page_at_lsn(
                     TESTREL_A,
                     1,
-                    Version::at(Lsn(0x60)),
+                    Version::Lsn(Lsn(0x60)),
                     &ctx,
                     io_concurrency.clone()
                 )
@@ -1854,7 +1854,7 @@ mod tests {
         // should still see the truncated block with older LSN
         assert_eq!(
             tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x50)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x50)), &ctx)
                 .await?,
             3
         );
@@ -1863,7 +1863,7 @@ mod tests {
                 .get_rel_page_at_lsn(
                     TESTREL_A,
                     2,
-                    Version::at(Lsn(0x50)),
+                    Version::Lsn(Lsn(0x50)),
                     &ctx,
                     io_concurrency.clone()
                 )
@@ -1880,7 +1880,7 @@ mod tests {
         m.commit(&ctx).await?;
         assert_eq!(
             tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x68)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x68)), &ctx)
                 .await?,
             0
         );
@@ -1893,7 +1893,7 @@ mod tests {
         m.commit(&ctx).await?;
         assert_eq!(
             tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x70)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x70)), &ctx)
                 .await?,
             2
         );
@@ -1902,7 +1902,7 @@ mod tests {
                 .get_rel_page_at_lsn(
                     TESTREL_A,
                     0,
-                    Version::at(Lsn(0x70)),
+                    Version::Lsn(Lsn(0x70)),
                     &ctx,
                     io_concurrency.clone()
                 )
@@ -1915,7 +1915,7 @@ mod tests {
                 .get_rel_page_at_lsn(
                     TESTREL_A,
                     1,
-                    Version::at(Lsn(0x70)),
+                    Version::Lsn(Lsn(0x70)),
                     &ctx,
                     io_concurrency.clone()
                 )
@@ -1932,7 +1932,7 @@ mod tests {
         m.commit(&ctx).await?;
         assert_eq!(
             tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x80)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x80)), &ctx)
                 .await?,
             1501
         );
@@ -1942,7 +1942,7 @@ mod tests {
                     .get_rel_page_at_lsn(
                         TESTREL_A,
                         blk,
-                        Version::at(Lsn(0x80)),
+                        Version::Lsn(Lsn(0x80)),
                         &ctx,
                         io_concurrency.clone()
                     )
@@ -1956,7 +1956,7 @@ mod tests {
                 .get_rel_page_at_lsn(
                     TESTREL_A,
                     1500,
-                    Version::at(Lsn(0x80)),
+                    Version::Lsn(Lsn(0x80)),
                     &ctx,
                     io_concurrency.clone()
                 )
@@ -1990,13 +1990,13 @@ mod tests {
         // Check that rel exists and size is correct
         assert_eq!(
             tline
-                .get_rel_exists(TESTREL_A, Version::at(Lsn(0x20)), &ctx)
+                .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x20)), &ctx)
                 .await?,
             true
         );
         assert_eq!(
             tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x20)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x20)), &ctx)
                 .await?,
             1
         );
@@ -2011,7 +2011,7 @@ mod tests {
         // Check that rel is not visible anymore
         assert_eq!(
             tline
-                .get_rel_exists(TESTREL_A, Version::at(Lsn(0x30)), &ctx)
+                .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x30)), &ctx)
                 .await?,
             false
         );
@@ -2029,13 +2029,13 @@ mod tests {
         // Check that rel exists and size is correct
         assert_eq!(
             tline
-                .get_rel_exists(TESTREL_A, Version::at(Lsn(0x40)), &ctx)
+                .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x40)), &ctx)
                 .await?,
             true
         );
         assert_eq!(
             tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x40)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x40)), &ctx)
                 .await?,
             1
         );
@@ -2077,26 +2077,26 @@ mod tests {
         // The relation was created at LSN 20, not visible at LSN 1 yet.
         assert_eq!(
             tline
-                .get_rel_exists(TESTREL_A, Version::at(Lsn(0x10)), &ctx)
+                .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x10)), &ctx)
                 .await?,
             false
         );
         assert!(
             tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x10)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x10)), &ctx)
                 .await
                 .is_err()
         );
 
         assert_eq!(
             tline
-                .get_rel_exists(TESTREL_A, Version::at(Lsn(0x20)), &ctx)
+                .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x20)), &ctx)
                 .await?,
             true
         );
         assert_eq!(
             tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x20)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x20)), &ctx)
                 .await?,
             relsize
         );
@@ -2110,7 +2110,7 @@ mod tests {
                     .get_rel_page_at_lsn(
                         TESTREL_A,
                         blkno,
-                        Version::at(lsn),
+                        Version::Lsn(lsn),
                         &ctx,
                         io_concurrency.clone()
                     )
@@ -2131,7 +2131,7 @@ mod tests {
         // Check reported size and contents after truncation
         assert_eq!(
             tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x60)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x60)), &ctx)
                 .await?,
             1
         );
@@ -2144,7 +2144,7 @@ mod tests {
                     .get_rel_page_at_lsn(
                         TESTREL_A,
                         blkno,
-                        Version::at(Lsn(0x60)),
+                        Version::Lsn(Lsn(0x60)),
                         &ctx,
                         io_concurrency.clone()
                     )
@@ -2157,7 +2157,7 @@ mod tests {
         // should still see all blocks with older LSN
         assert_eq!(
             tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x50)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x50)), &ctx)
                 .await?,
             relsize
         );
@@ -2169,7 +2169,7 @@ mod tests {
                     .get_rel_page_at_lsn(
                         TESTREL_A,
                         blkno,
-                        Version::at(Lsn(0x50)),
+                        Version::Lsn(Lsn(0x50)),
                         &ctx,
                         io_concurrency.clone()
                     )
@@ -2193,13 +2193,13 @@ mod tests {
 
         assert_eq!(
             tline
-                .get_rel_exists(TESTREL_A, Version::at(Lsn(0x80)), &ctx)
+                .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x80)), &ctx)
                 .await?,
             true
         );
         assert_eq!(
             tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(0x80)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x80)), &ctx)
                 .await?,
             relsize
         );
@@ -2212,7 +2212,7 @@ mod tests {
                     .get_rel_page_at_lsn(
                         TESTREL_A,
                         blkno,
-                        Version::at(Lsn(0x80)),
+                        Version::Lsn(Lsn(0x80)),
                         &ctx,
                         io_concurrency.clone()
                     )
@@ -2250,7 +2250,7 @@ mod tests {
 
         assert_eq!(
             tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(lsn)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(lsn)), &ctx)
                 .await?,
             RELSEG_SIZE + 1
         );
@@ -2264,7 +2264,7 @@ mod tests {
         m.commit(&ctx).await?;
         assert_eq!(
             tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(lsn)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(lsn)), &ctx)
                 .await?,
             RELSEG_SIZE
         );
@@ -2279,7 +2279,7 @@ mod tests {
         m.commit(&ctx).await?;
         assert_eq!(
             tline
-                .get_rel_size(TESTREL_A, Version::at(Lsn(lsn)), &ctx)
+                .get_rel_size(TESTREL_A, Version::Lsn(Lsn(lsn)), &ctx)
                 .await?,
             RELSEG_SIZE - 1
         );
@@ -2297,7 +2297,7 @@ mod tests {
             m.commit(&ctx).await?;
             assert_eq!(
                 tline
-                    .get_rel_size(TESTREL_A, Version::at(Lsn(lsn)), &ctx)
+                    .get_rel_size(TESTREL_A, Version::Lsn(Lsn(lsn)), &ctx)
                     .await?,
                 size as BlockNumber
             );

proxy/src/auth/backend/mod.rs

@@ -536,8 +536,7 @@ mod tests {
     use control_plane::AuthSecret;
     use fallible_iterator::FallibleIterator;
     use once_cell::sync::Lazy;
-    use postgres_protocol::CSafeStr;
-    use postgres_protocol::authentication::sasl::{ChannelBinding, SCRAM_SHA_256, ScramSha256};
+    use postgres_protocol::authentication::sasl::{ChannelBinding, ScramSha256};
     use postgres_protocol::message::backend::Message as PgMessage;
     use postgres_protocol::message::frontend;
     use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt};
@@ -715,15 +714,15 @@ mod tests {
             // server should offer scram
             match read_message(&mut client, &mut read).await {
                 PgMessage::AuthenticationSasl(a) => {
-                    let options: Vec<&CSafeStr> = a.mechanisms().collect().unwrap();
-                    assert_eq!(options, [SCRAM_SHA_256]);
+                    let options: Vec<&str> = a.mechanisms().collect().unwrap();
+                    assert_eq!(options, ["SCRAM-SHA-256"]);
                 }
                 _ => panic!("wrong message"),
             }
 
             // client sends client-first-message
             let mut write = BytesMut::new();
-            frontend::sasl_initial_response(SCRAM_SHA_256, scram.message(), &mut write);
+            frontend::sasl_initial_response("SCRAM-SHA-256", scram.message(), &mut write).unwrap();
             client.write_all(&write).await.unwrap();
 
             // server response with server-first-message
@@ -736,7 +735,7 @@ mod tests {
 
             // client response with client-final-message
             write.clear();
-            frontend::sasl_response(scram.message(), &mut write);
+            frontend::sasl_response(scram.message(), &mut write).unwrap();
             client.write_all(&write).await.unwrap();
 
             // server response with server-final-message
@@ -801,7 +800,7 @@ mod tests {
 
             // client responds with password
             write.clear();
-            frontend::password_message(c"my-secret-password".into(), &mut write);
+            frontend::password_message(b"my-secret-password", &mut write).unwrap();
             client.write_all(&write).await.unwrap();
         });
         let endpoint_rate_limiter = Arc::new(EndpointRateLimiter::new_with_shards(
@@ -854,10 +853,8 @@ mod tests {
 
             // client responds with password
             let mut write = BytesMut::new();
-            frontend::password_message(
-                c"endpoint=my-endpoint;my-secret-password".into(),
-                &mut write,
-            );
+            frontend::password_message(b"endpoint=my-endpoint;my-secret-password", &mut write)
+                .unwrap();
             client.write_all(&write).await.unwrap();
         });
 

proxy/src/auth/flow.rs

@@ -3,6 +3,7 @@
 use std::io;
 use std::sync::Arc;
 
+use postgres_protocol::authentication::sasl::{SCRAM_SHA_256, SCRAM_SHA_256_PLUS};
 use pq_proto::{BeAuthenticationSaslMessage, BeMessage, BeMessage as Be};
 use tokio::io::{AsyncRead, AsyncWrite};
 use tracing::info;
@@ -173,10 +174,8 @@ impl<S: AsyncRead + AsyncWrite + Unpin> AuthFlow<'_, S, Scram<'_>> {
         }
 
         match sasl.method {
-            scram::SCRAM_SHA_256 => ctx.set_auth_method(crate::context::AuthMethod::ScramSha256),
-            scram::SCRAM_SHA_256_PLUS => {
-                ctx.set_auth_method(crate::context::AuthMethod::ScramSha256Plus)
-            }
+            SCRAM_SHA_256 => ctx.set_auth_method(crate::context::AuthMethod::ScramSha256),
+            SCRAM_SHA_256_PLUS => ctx.set_auth_method(crate::context::AuthMethod::ScramSha256Plus),
             _ => {}
         }
 

proxy/src/binary/proxy.rs

@@ -161,11 +161,8 @@ struct ProxyCliArgs {
     #[clap(long, default_values_t = RateBucketInfo::DEFAULT_REDIS_SET)]
     redis_rps_limit: Vec<RateBucketInfo>,
     /// Cancellation channel size (max queue size for redis kv client)
-    #[clap(long, default_value_t = 1024)]
+    #[clap(long, default_value = "1024")]
     cancellation_ch_size: usize,
-    /// Cancellation ops batch size for redis
-    #[clap(long, default_value_t = 8)]
-    cancellation_batch_size: usize,
     /// cache for `allowed_ips` (use `size=0` to disable)
     #[clap(long, default_value = config::CacheOptions::CACHE_DEFAULT_OPTIONS)]
     allowed_ips_cache: String,
@@ -545,12 +542,7 @@ pub async fn run() -> anyhow::Result<()> {
             if let Some(mut redis_kv_client) = redis_kv_client {
                 maintenance_tasks.spawn(async move {
                     redis_kv_client.try_connect().await?;
-                    handle_cancel_messages(
-                        &mut redis_kv_client,
-                        rx_cancel,
-                        args.cancellation_batch_size,
-                    )
-                    .await?;
+                    handle_cancel_messages(&mut redis_kv_client, rx_cancel).await?;
 
                     drop(redis_kv_client);
 

proxy/src/cancellation.rs

@@ -30,6 +30,8 @@ use crate::tls::postgres_rustls::MakeRustlsConnect;
 type IpSubnetKey = IpNet;
 
 const CANCEL_KEY_TTL: i64 = 1_209_600; // 2 weeks cancellation key expire time
+const REDIS_SEND_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(10);
+const BATCH_SIZE: usize = 8;
 
 // Message types for sending through mpsc channel
 pub enum CancelKeyOp {
@@ -229,13 +231,12 @@ impl CancelReplyOp {
 pub async fn handle_cancel_messages(
     client: &mut RedisKVClient,
     mut rx: mpsc::Receiver<CancelKeyOp>,
-    batch_size: usize,
 ) -> anyhow::Result<()> {
-    let mut batch = Vec::with_capacity(batch_size);
-    let mut pipeline = Pipeline::with_capacity(batch_size);
+    let mut batch = Vec::with_capacity(BATCH_SIZE);
+    let mut pipeline = Pipeline::with_capacity(BATCH_SIZE);
 
     loop {
-        if rx.recv_many(&mut batch, batch_size).await == 0 {
+        if rx.recv_many(&mut batch, BATCH_SIZE).await == 0 {
             warn!("shutting down cancellation queue");
             break Ok(());
         }
@@ -366,7 +367,8 @@ impl CancellationHandler {
             return Err(CancelError::InternalError);
         };
 
-        tx.try_send(op)
+        tx.send_timeout(op, REDIS_SEND_TIMEOUT)
+            .await
             .map_err(|e| {
                 tracing::warn!("failed to send GetCancelData for {key}: {e}");
             })
@@ -568,7 +570,7 @@ impl Session {
     }
 
     // Send the store key op to the cancellation handler and set TTL for the key
-    pub(crate) fn write_cancel_key(
+    pub(crate) async fn write_cancel_key(
         &self,
         cancel_closure: CancelClosure,
     ) -> Result<(), CancelError> {
@@ -594,14 +596,14 @@ impl Session {
             expire: CANCEL_KEY_TTL,
         };
 
-        let _ = tx.try_send(op).map_err(|e| {
+        let _ = tx.send_timeout(op, REDIS_SEND_TIMEOUT).await.map_err(|e| {
             let key = self.key;
             tracing::warn!("failed to send StoreCancelKey for {key}: {e}");
         });
         Ok(())
     }
 
-    pub(crate) fn remove_cancel_key(&self) -> Result<(), CancelError> {
+    pub(crate) async fn remove_cancel_key(&self) -> Result<(), CancelError> {
         let Some(tx) = &self.cancellation_handler.tx else {
             tracing::warn!("cancellation handler is not available");
             return Err(CancelError::InternalError);
@@ -617,7 +619,7 @@ impl Session {
                 .guard(RedisMsgKind::HDel),
         };
 
-        let _ = tx.try_send(op).map_err(|e| {
+        let _ = tx.send_timeout(op, REDIS_SEND_TIMEOUT).await.map_err(|e| {
             let key = self.key;
             tracing::warn!("failed to send RemoveCancelKey for {key}: {e}");
         });

proxy/src/console_redirect_proxy.rs

@@ -244,7 +244,9 @@ pub(crate) async fn handle_client<S: AsyncRead + AsyncWrite + Unpin>(
     let cancellation_handler_clone = Arc::clone(&cancellation_handler);
     let session = cancellation_handler_clone.get_key();
 
-    session.write_cancel_key(node.cancel_closure.clone())?;
+    session
+        .write_cancel_key(node.cancel_closure.clone())
+        .await?;
 
     prepare_client_connection(&node, *session.key(), &mut stream).await?;
 

proxy/src/logging.rs

@@ -1,11 +1,13 @@
-use std::cell::RefCell;
+use std::cell::{Cell, RefCell};
 use std::collections::HashMap;
-use std::sync::Arc;
+use std::hash::BuildHasher;
 use std::sync::atomic::{AtomicU32, Ordering};
-use std::{env, io};
+use std::{array, env, fmt, io};
 
 use chrono::{DateTime, Utc};
+use indexmap::IndexSet;
 use opentelemetry::trace::TraceContextExt;
+use scopeguard::defer;
 use serde::ser::{SerializeMap, Serializer};
 use tracing::subscriber::Interest;
 use tracing::{Event, Metadata, Span, Subscriber, callsite, span};
@@ -17,6 +19,7 @@ use tracing_subscriber::fmt::{FormatEvent, FormatFields};
 use tracing_subscriber::layer::{Context, Layer};
 use tracing_subscriber::prelude::*;
 use tracing_subscriber::registry::{LookupSpan, SpanRef};
+use try_lock::TryLock;
 
 /// Initialize logging and OpenTelemetry tracing and exporter.
 ///
@@ -52,7 +55,7 @@ pub async fn init() -> anyhow::Result<LoggingGuard> {
             StderrWriter {
                 stderr: std::io::stderr(),
             },
-            &["request_id", "session_id", "conn_id"],
+            ["request_id", "session_id", "conn_id"],
         ))
     } else {
         None
@@ -180,65 +183,50 @@ impl Clock for RealClock {
 /// Name of the field used by tracing crate to store the event message.
 const MESSAGE_FIELD: &str = "message";
 
-/// Tracing used to enforce that spans/events have no more than 32 fields.
-/// It seems this is no longer the case, but it's still documented in some places.
-/// Generally, we shouldn't expect more than 32 fields anyway, so we can try and
-/// rely on it for some (minor) performance gains.
-const MAX_TRACING_FIELDS: usize = 32;
-
 thread_local! {
+    /// Protects against deadlocks and double panics during log writing.
+    /// The current panic handler will use tracing to log panic information.
+    static REENTRANCY_GUARD: Cell<bool> = const { Cell::new(false) };
     /// Thread-local instance with per-thread buffer for log writing.
-    static EVENT_FORMATTER: RefCell<EventFormatter> = const { RefCell::new(EventFormatter::new()) };
+    static EVENT_FORMATTER: RefCell<EventFormatter> = RefCell::new(EventFormatter::new());
     /// Cached OS thread ID.
     static THREAD_ID: u64 = gettid::gettid();
 }
 
-/// Map for values fixed at callsite registration.
-// We use papaya here because registration rarely happens post-startup.
-// papaya is good for read-heavy workloads.
-//
-// We use rustc_hash here because callsite::Identifier will always be an integer with low-bit entropy,
-// since it's always a pointer to static mutable data. rustc_hash was designed for low-bit entropy.
-type CallsiteMap<T> =
-    papaya::HashMap<callsite::Identifier, T, std::hash::BuildHasherDefault<rustc_hash::FxHasher>>;
-
 /// Implements tracing layer to handle events specific to logging.
-struct JsonLoggingLayer<C: Clock, W: MakeWriter> {
+struct JsonLoggingLayer<C: Clock, W: MakeWriter, const F: usize> {
     clock: C,
+    skipped_field_indices: papaya::HashMap<callsite::Identifier, SkippedFieldIndices>,
+    callsite_ids: papaya::HashMap<callsite::Identifier, CallsiteId>,
     writer: W,
-
-    /// tracks which fields of each **event** are duplicates
-    skipped_field_indices: CallsiteMap<SkippedFieldIndices>,
-
-    span_info: CallsiteMap<CallsiteSpanInfo>,
-
-    /// Fields we want to keep track of in a separate json object.
-    extract_fields: &'static [&'static str],
+    // We use a const generic and arrays to bypass one heap allocation.
+    extract_fields: IndexSet<&'static str>,
+    _marker: std::marker::PhantomData<[&'static str; F]>,
 }
 
-impl<C: Clock, W: MakeWriter> JsonLoggingLayer<C, W> {
-    fn new(clock: C, writer: W, extract_fields: &'static [&'static str]) -> Self {
+impl<C: Clock, W: MakeWriter, const F: usize> JsonLoggingLayer<C, W, F> {
+    fn new(clock: C, writer: W, extract_fields: [&'static str; F]) -> Self {
         JsonLoggingLayer {
             clock,
-            skipped_field_indices: CallsiteMap::default(),
-            span_info: CallsiteMap::default(),
+            skipped_field_indices: papaya::HashMap::default(),
+            callsite_ids: papaya::HashMap::default(),
             writer,
-            extract_fields,
+            extract_fields: IndexSet::from_iter(extract_fields),
+            _marker: std::marker::PhantomData,
         }
     }
 
     #[inline]
-    fn span_info(&self, metadata: &'static Metadata<'static>) -> CallsiteSpanInfo {
-        self.span_info
+    fn callsite_id(&self, cs: callsite::Identifier) -> CallsiteId {
+        *self
+            .callsite_ids
             .pin()
-            .get_or_insert_with(metadata.callsite(), || {
-                CallsiteSpanInfo::new(metadata, self.extract_fields)
-            })
-            .clone()
+            .get_or_insert_with(cs, CallsiteId::next)
     }
 }
 
-impl<S, C: Clock + 'static, W: MakeWriter + 'static> Layer<S> for JsonLoggingLayer<C, W>
+impl<S, C: Clock + 'static, W: MakeWriter + 'static, const F: usize> Layer<S>
+    for JsonLoggingLayer<C, W, F>
 where
     S: Subscriber + for<'a> LookupSpan<'a>,
 {
@@ -249,25 +237,35 @@ where
         //       early, before OTel machinery, and add as event extension.
         let now = self.clock.now();
 
-        let res: io::Result<()> = EVENT_FORMATTER.with(|f| {
-            let mut borrow = f.try_borrow_mut();
-            let formatter = match borrow.as_deref_mut() {
-                Ok(formatter) => formatter,
-                // If the thread local formatter is borrowed,
-                // then we likely hit an edge case were we panicked during formatting.
-                // We allow the logging to proceed with an uncached formatter.
-                Err(_) => &mut EventFormatter::new(),
-            };
+        let res: io::Result<()> = REENTRANCY_GUARD.with(move |entered| {
+            if entered.get() {
+                let mut formatter = EventFormatter::new();
+                formatter.format::<S, F>(
+                    now,
+                    event,
+                    &ctx,
+                    &self.skipped_field_indices,
+                    &self.callsite_ids,
+                    &self.extract_fields,
+                )?;
+                self.writer.make_writer().write_all(formatter.buffer())
+            } else {
+                entered.set(true);
+                defer!(entered.set(false););
 
-            formatter.reset();
-            formatter.format(
-                now,
-                event,
-                &ctx,
-                &self.skipped_field_indices,
-                self.extract_fields,
-            )?;
-            self.writer.make_writer().write_all(formatter.buffer())
+                EVENT_FORMATTER.with_borrow_mut(move |formatter| {
+                    formatter.reset();
+                    formatter.format::<S, F>(
+                        now,
+                        event,
+                        &ctx,
+                        &self.skipped_field_indices,
+                        &self.callsite_ids,
+                        &self.extract_fields,
+                    )?;
+                    self.writer.make_writer().write_all(formatter.buffer())
+                })
+            }
         });
 
         // In case logging fails we generate a simpler JSON object.
@@ -289,48 +287,50 @@ where
     /// Registers a SpanFields instance as span extension.
     fn on_new_span(&self, attrs: &span::Attributes<'_>, id: &span::Id, ctx: Context<'_, S>) {
         let span = ctx.span(id).expect("span must exist");
+        let fields = SpanFields::default();
+        fields.record_fields(attrs);
 
-        let mut fields = SpanFields::new(self.span_info(span.metadata()));
-        attrs.record(&mut fields);
+        // This could deadlock when there's a panic somewhere in the tracing
+        // event handling and a read or write guard is still held. This includes
+        // the OTel subscriber.
+        let mut exts = span.extensions_mut();
 
-        // This is a new span: the extensions should not be locked
-        // unless some layer spawned a thread to process this span.
-        // I don't think any layers do that.
-        span.extensions_mut().insert(fields);
+        exts.insert(fields);
     }
 
     fn on_record(&self, id: &span::Id, values: &span::Record<'_>, ctx: Context<'_, S>) {
         let span = ctx.span(id).expect("span must exist");
-
-        // assumption: `on_record` is rarely called.
-        // assumption: a span being updated by one thread,
-        //             and formatted by another thread is even rarer.
-        let mut ext = span.extensions_mut();
-        if let Some(fields) = ext.get_mut::<SpanFields>() {
-            values.record(fields);
+        let ext = span.extensions();
+        if let Some(data) = ext.get::<SpanFields>() {
+            data.record_fields(values);
         }
     }
 
-    /// Called (lazily) roughly once per event/span instance. We quickly check
-    /// for duplicate field names and record duplicates as skippable. Last field wins.
+    /// Called (lazily) whenever a new log call is executed. We quickly check
+    /// for duplicate field names and record duplicates as skippable. Last one
+    /// wins.
     fn register_callsite(&self, metadata: &'static Metadata<'static>) -> Interest {
-        debug_assert!(
-            metadata.fields().len() <= MAX_TRACING_FIELDS,
-            "callsite {metadata:?} has too many fields."
-        );
-
         if !metadata.is_event() {
-            // register the span info.
-            self.span_info(metadata);
+            self.callsite_id(metadata.callsite());
             // Must not be never because we wouldn't get trace and span data.
             return Interest::always();
         }
 
         let mut field_indices = SkippedFieldIndices::default();
-        let mut seen_fields = HashMap::new();
+        let mut seen_fields = HashMap::<&'static str, usize>::new();
         for field in metadata.fields() {
-            if let Some(old_index) = seen_fields.insert(field.name(), field.index()) {
-                field_indices.set(old_index);
+            use std::collections::hash_map::Entry;
+            match seen_fields.entry(field.name()) {
+                Entry::Vacant(entry) => {
+                    // field not seen yet
+                    entry.insert(field.index());
+                }
+                Entry::Occupied(mut entry) => {
+                    // replace currently stored index
+                    let old_index = entry.insert(field.index());
+                    // ... and append it to list of skippable indices
+                    field_indices.push(old_index);
+                }
             }
         }
 
@@ -344,113 +344,110 @@ where
     }
 }
 
-/// Any span info that is fixed to a particular callsite. Not variable between span instances.
-#[derive(Clone)]
-struct CallsiteSpanInfo {
-    /// index of each field to extract. usize::MAX if not found.
-    extract: Arc<[usize]>,
+#[derive(Copy, Clone, Debug, Default)]
+#[repr(transparent)]
+struct CallsiteId(u32);
 
-    /// tracks the fixed "callsite ID" for each span.
-    /// note: this is not stable between runs.
-    normalized_name: Arc<str>,
-}
-
-impl CallsiteSpanInfo {
-    fn new(metadata: &'static Metadata<'static>, extract_fields: &[&'static str]) -> Self {
+impl CallsiteId {
+    #[inline]
+    fn next() -> Self {
         // Start at 1 to reserve 0 for default.
         static COUNTER: AtomicU32 = AtomicU32::new(1);
+        CallsiteId(COUNTER.fetch_add(1, Ordering::Relaxed))
+    }
+}
 
-        let names: Vec<&'static str> = metadata.fields().iter().map(|f| f.name()).collect();
-
-        // get all the indices of span fields we want to focus
-        let extract = extract_fields
-            .iter()
-            // use rposition, since we want last match wins.
-            .map(|f1| names.iter().rposition(|f2| f1 == f2).unwrap_or(usize::MAX))
-            .collect();
-
-        // normalized_name is unique for each callsite, but it is not
-        // unified across separate proxy instances.
-        // todo: can we do better here?
-        let cid = COUNTER.fetch_add(1, Ordering::Relaxed);
-        let normalized_name = format!("{}#{cid}", metadata.name()).into();
-
-        Self {
-            extract,
-            normalized_name,
-        }
+impl fmt::Display for CallsiteId {
+    #[inline]
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.0.fmt(f)
     }
 }
 
 /// Stores span field values recorded during the spans lifetime.
+#[derive(Default)]
 struct SpanFields {
-    values: [serde_json::Value; MAX_TRACING_FIELDS],
-
-    /// cached span info so we can avoid extra hashmap lookups in the hot path.
-    span_info: CallsiteSpanInfo,
+    // TODO: Switch to custom enum with lasso::Spur for Strings?
+    fields: papaya::HashMap<&'static str, serde_json::Value>,
 }
 
 impl SpanFields {
-    fn new(span_info: CallsiteSpanInfo) -> Self {
-        Self {
-            span_info,
-            values: [const { serde_json::Value::Null }; MAX_TRACING_FIELDS],
-        }
+    #[inline]
+    fn record_fields<R: tracing_subscriber::field::RecordFields>(&self, fields: R) {
+        fields.record(&mut SpanFieldsRecorder {
+            fields: self.fields.pin(),
+        });
     }
 }
 
-impl tracing::field::Visit for SpanFields {
+/// Implements a tracing field visitor to convert and store values.
+struct SpanFieldsRecorder<'m, S, G> {
+    fields: papaya::HashMapRef<'m, &'static str, serde_json::Value, S, G>,
+}
+
+impl<S: BuildHasher, G: papaya::Guard> tracing::field::Visit for SpanFieldsRecorder<'_, S, G> {
     #[inline]
     fn record_f64(&mut self, field: &tracing::field::Field, value: f64) {
-        self.values[field.index()] = serde_json::Value::from(value);
+        self.fields
+            .insert(field.name(), serde_json::Value::from(value));
     }
 
     #[inline]
     fn record_i64(&mut self, field: &tracing::field::Field, value: i64) {
-        self.values[field.index()] = serde_json::Value::from(value);
+        self.fields
+            .insert(field.name(), serde_json::Value::from(value));
     }
 
     #[inline]
     fn record_u64(&mut self, field: &tracing::field::Field, value: u64) {
-        self.values[field.index()] = serde_json::Value::from(value);
+        self.fields
+            .insert(field.name(), serde_json::Value::from(value));
     }
 
     #[inline]
     fn record_i128(&mut self, field: &tracing::field::Field, value: i128) {
         if let Ok(value) = i64::try_from(value) {
-            self.values[field.index()] = serde_json::Value::from(value);
+            self.fields
+                .insert(field.name(), serde_json::Value::from(value));
         } else {
-            self.values[field.index()] = serde_json::Value::from(format!("{value}"));
+            self.fields
+                .insert(field.name(), serde_json::Value::from(format!("{value}")));
         }
     }
 
     #[inline]
     fn record_u128(&mut self, field: &tracing::field::Field, value: u128) {
         if let Ok(value) = u64::try_from(value) {
-            self.values[field.index()] = serde_json::Value::from(value);
+            self.fields
+                .insert(field.name(), serde_json::Value::from(value));
         } else {
-            self.values[field.index()] = serde_json::Value::from(format!("{value}"));
+            self.fields
+                .insert(field.name(), serde_json::Value::from(format!("{value}")));
         }
     }
 
     #[inline]
     fn record_bool(&mut self, field: &tracing::field::Field, value: bool) {
-        self.values[field.index()] = serde_json::Value::from(value);
+        self.fields
+            .insert(field.name(), serde_json::Value::from(value));
     }
 
     #[inline]
     fn record_bytes(&mut self, field: &tracing::field::Field, value: &[u8]) {
-        self.values[field.index()] = serde_json::Value::from(value);
+        self.fields
+            .insert(field.name(), serde_json::Value::from(value));
     }
 
     #[inline]
     fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
-        self.values[field.index()] = serde_json::Value::from(value);
+        self.fields
+            .insert(field.name(), serde_json::Value::from(value));
     }
 
     #[inline]
     fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
-        self.values[field.index()] = serde_json::Value::from(format!("{value:?}"));
+        self.fields
+            .insert(field.name(), serde_json::Value::from(format!("{value:?}")));
     }
 
     #[inline]
@@ -459,33 +456,38 @@ impl tracing::field::Visit for SpanFields {
         field: &tracing::field::Field,
         value: &(dyn std::error::Error + 'static),
     ) {
-        self.values[field.index()] = serde_json::Value::from(format!("{value}"));
+        self.fields
+            .insert(field.name(), serde_json::Value::from(format!("{value}")));
     }
 }
 
 /// List of field indices skipped during logging. Can list duplicate fields or
 /// metafields not meant to be logged.
-#[derive(Copy, Clone, Default)]
+#[derive(Clone, Default)]
 struct SkippedFieldIndices {
-    // 32-bits is large enough for `MAX_TRACING_FIELDS`
-    bits: u32,
+    bits: u64,
 }
 
 impl SkippedFieldIndices {
     #[inline]
-    fn is_empty(self) -> bool {
+    fn is_empty(&self) -> bool {
         self.bits == 0
     }
 
     #[inline]
-    fn set(&mut self, index: usize) {
-        debug_assert!(index <= 32, "index out of bounds of 32-bit set");
-        self.bits |= 1 << index;
+    fn push(&mut self, index: usize) {
+        self.bits |= 1u64
+            .checked_shl(index as u32)
+            .expect("field index too large");
     }
 
     #[inline]
-    fn contains(self, index: usize) -> bool {
-        self.bits & (1 << index) != 0
+    fn contains(&self, index: usize) -> bool {
+        self.bits
+            & 1u64
+                .checked_shl(index as u32)
+                .expect("field index too large")
+            != 0
     }
 }
 
@@ -497,7 +499,7 @@ struct EventFormatter {
 
 impl EventFormatter {
     #[inline]
-    const fn new() -> Self {
+    fn new() -> Self {
         EventFormatter {
             logline_buffer: Vec::new(),
         }
@@ -513,13 +515,14 @@ impl EventFormatter {
         self.logline_buffer.clear();
     }
 
-    fn format<S>(
+    fn format<S, const F: usize>(
         &mut self,
         now: DateTime<Utc>,
         event: &Event<'_>,
         ctx: &Context<'_, S>,
-        skipped_field_indices: &CallsiteMap<SkippedFieldIndices>,
-        extract_fields: &'static [&'static str],
+        skipped_field_indices: &papaya::HashMap<callsite::Identifier, SkippedFieldIndices>,
+        callsite_ids: &papaya::HashMap<callsite::Identifier, CallsiteId>,
+        extract_fields: &IndexSet<&'static str>,
     ) -> io::Result<()>
     where
         S: Subscriber + for<'a> LookupSpan<'a>,
@@ -530,11 +533,8 @@ impl EventFormatter {
         let normalized_meta = event.normalized_metadata();
         let meta = normalized_meta.as_ref().unwrap_or_else(|| event.metadata());
 
-        let skipped_field_indices = skipped_field_indices
-            .pin()
-            .get(&meta.callsite())
-            .copied()
-            .unwrap_or_default();
+        let skipped_field_indices = skipped_field_indices.pin();
+        let skipped_field_indices = skipped_field_indices.get(&meta.callsite());
 
         let mut serialize = || {
             let mut serializer = serde_json::Serializer::new(&mut self.logline_buffer);
@@ -565,11 +565,9 @@ impl EventFormatter {
             }
 
             let spans = SerializableSpans {
-                // collect all spans from parent to root.
-                spans: ctx
-                    .event_span(event)
-                    .map_or(vec![], |parent| parent.scope().collect()),
-                extracted: ExtractedSpanFields::new(extract_fields),
+                ctx,
+                callsite_ids,
+                extract: ExtractedSpanFields::<'_, F>::new(extract_fields),
             };
             serializer.serialize_entry("spans", &spans)?;
 
@@ -622,9 +620,9 @@ impl EventFormatter {
                 }
             }
 
-            if spans.extracted.has_values() {
+            if spans.extract.has_values() {
                 // TODO: add fields from event, too?
-                serializer.serialize_entry("extract", &spans.extracted)?;
+                serializer.serialize_entry("extract", &spans.extract)?;
             }
 
             serializer.end()
@@ -637,15 +635,15 @@ impl EventFormatter {
 }
 
 /// Extracts the message field that's mixed will other fields.
-struct MessageFieldExtractor<S: serde::ser::SerializeMap> {
+struct MessageFieldExtractor<'a, S: serde::ser::SerializeMap> {
     serializer: S,
-    skipped_field_indices: SkippedFieldIndices,
+    skipped_field_indices: Option<&'a SkippedFieldIndices>,
     state: Option<Result<(), S::Error>>,
 }
 
-impl<S: serde::ser::SerializeMap> MessageFieldExtractor<S> {
+impl<'a, S: serde::ser::SerializeMap> MessageFieldExtractor<'a, S> {
     #[inline]
-    fn new(serializer: S, skipped_field_indices: SkippedFieldIndices) -> Self {
+    fn new(serializer: S, skipped_field_indices: Option<&'a SkippedFieldIndices>) -> Self {
         Self {
             serializer,
             skipped_field_indices,
@@ -667,11 +665,13 @@ impl<S: serde::ser::SerializeMap> MessageFieldExtractor<S> {
     fn accept_field(&self, field: &tracing::field::Field) -> bool {
         self.state.is_none()
             && field.name() == MESSAGE_FIELD
-            && !self.skipped_field_indices.contains(field.index())
+            && !self
+                .skipped_field_indices
+                .is_some_and(|i| i.contains(field.index()))
     }
 }
 
-impl<S: serde::ser::SerializeMap> tracing::field::Visit for MessageFieldExtractor<S> {
+impl<S: serde::ser::SerializeMap> tracing::field::Visit for MessageFieldExtractor<'_, S> {
     #[inline]
     fn record_f64(&mut self, field: &tracing::field::Field, value: f64) {
         if self.accept_field(field) {
@@ -751,14 +751,14 @@ impl<S: serde::ser::SerializeMap> tracing::field::Visit for MessageFieldExtracto
 /// can be skipped.
 // This is entirely optional and only cosmetic, though maybe helps a
 // bit during log parsing in dashboards when there's no field with empty object.
-struct FieldsPresent(pub bool, SkippedFieldIndices);
+struct FieldsPresent<'a>(pub bool, Option<&'a SkippedFieldIndices>);
 
 // Even though some methods have an overhead (error, bytes) it is assumed the
 // compiler won't include this since we ignore the value entirely.
-impl tracing::field::Visit for FieldsPresent {
+impl tracing::field::Visit for FieldsPresent<'_> {
     #[inline]
     fn record_debug(&mut self, field: &tracing::field::Field, _: &dyn std::fmt::Debug) {
-        if !self.1.contains(field.index())
+        if !self.1.is_some_and(|i| i.contains(field.index()))
             && field.name() != MESSAGE_FIELD
             && !field.name().starts_with("log.")
         {
@@ -768,7 +768,10 @@ impl tracing::field::Visit for FieldsPresent {
 }
 
 /// Serializes the fields directly supplied with a log event.
-struct SerializableEventFields<'a, 'event>(&'a tracing::Event<'event>, SkippedFieldIndices);
+struct SerializableEventFields<'a, 'event>(
+    &'a tracing::Event<'event>,
+    Option<&'a SkippedFieldIndices>,
+);
 
 impl serde::ser::Serialize for SerializableEventFields<'_, '_> {
     fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
@@ -785,15 +788,15 @@ impl serde::ser::Serialize for SerializableEventFields<'_, '_> {
 }
 
 /// A tracing field visitor that skips the message field.
-struct MessageFieldSkipper<S: serde::ser::SerializeMap> {
+struct MessageFieldSkipper<'a, S: serde::ser::SerializeMap> {
     serializer: S,
-    skipped_field_indices: SkippedFieldIndices,
+    skipped_field_indices: Option<&'a SkippedFieldIndices>,
     state: Result<(), S::Error>,
 }
 
-impl<S: serde::ser::SerializeMap> MessageFieldSkipper<S> {
+impl<'a, S: serde::ser::SerializeMap> MessageFieldSkipper<'a, S> {
     #[inline]
-    fn new(serializer: S, skipped_field_indices: SkippedFieldIndices) -> Self {
+    fn new(serializer: S, skipped_field_indices: Option<&'a SkippedFieldIndices>) -> Self {
         Self {
             serializer,
             skipped_field_indices,
@@ -806,7 +809,9 @@ impl<S: serde::ser::SerializeMap> MessageFieldSkipper<S> {
         self.state.is_ok()
             && field.name() != MESSAGE_FIELD
             && !field.name().starts_with("log.")
-            && !self.skipped_field_indices.contains(field.index())
+            && !self
+                .skipped_field_indices
+                .is_some_and(|i| i.contains(field.index()))
     }
 
     #[inline]
@@ -816,7 +821,7 @@ impl<S: serde::ser::SerializeMap> MessageFieldSkipper<S> {
     }
 }
 
-impl<S: serde::ser::SerializeMap> tracing::field::Visit for MessageFieldSkipper<S> {
+impl<S: serde::ser::SerializeMap> tracing::field::Visit for MessageFieldSkipper<'_, S> {
     #[inline]
     fn record_f64(&mut self, field: &tracing::field::Field, value: f64) {
         if self.accept_field(field) {
@@ -900,17 +905,18 @@ impl<S: serde::ser::SerializeMap> tracing::field::Visit for MessageFieldSkipper<
 /// with the span names as keys. To prevent collision we append a numberic value
 /// to the name. Also, collects any span fields we're interested in. Last one
 /// wins.
-struct SerializableSpans<'ctx, S>
+struct SerializableSpans<'a, 'ctx, Span, const F: usize>
 where
-    S: for<'lookup> LookupSpan<'lookup>,
+    Span: Subscriber + for<'lookup> LookupSpan<'lookup>,
 {
-    spans: Vec<SpanRef<'ctx, S>>,
-    extracted: ExtractedSpanFields,
+    ctx: &'a Context<'ctx, Span>,
+    callsite_ids: &'a papaya::HashMap<callsite::Identifier, CallsiteId>,
+    extract: ExtractedSpanFields<'a, F>,
 }
 
-impl<S> serde::ser::Serialize for SerializableSpans<'_, S>
+impl<Span, const F: usize> serde::ser::Serialize for SerializableSpans<'_, '_, Span, F>
 where
-    S: for<'lookup> LookupSpan<'lookup>,
+    Span: Subscriber + for<'lookup> LookupSpan<'lookup>,
 {
     fn serialize<Ser>(&self, serializer: Ser) -> Result<Ser::Ok, Ser::Error>
     where
@@ -918,22 +924,25 @@ where
     {
         let mut serializer = serializer.serialize_map(None)?;
 
-        for span in self.spans.iter().rev() {
-            let ext = span.extensions();
+        if let Some(leaf_span) = self.ctx.lookup_current() {
+            for span in leaf_span.scope().from_root() {
+                // Append a numeric callsite ID to the span name to keep the name unique
+                // in the JSON object.
+                let cid = self
+                    .callsite_ids
+                    .pin()
+                    .get(&span.metadata().callsite())
+                    .copied()
+                    .unwrap_or_default();
 
-            // all spans should have this extension.
-            let Some(fields) = ext.get() else { continue };
+                // Loki turns the # into an underscore during field name concatenation.
+                serializer.serialize_key(&format_args!("{}#{}", span.metadata().name(), &cid))?;
 
-            self.extracted.layer_span(fields);
-
-            let SpanFields { values, span_info } = fields;
-            serializer.serialize_entry(
-                &*span_info.normalized_name,
-                &SerializableSpanFields {
-                    fields: span.metadata().fields(),
-                    values,
-                },
-            )?;
+                serializer.serialize_value(&SerializableSpanFields {
+                    span: &span,
+                    extract: &self.extract,
+                })?;
+            }
         }
 
         serializer.end()
@@ -941,77 +950,80 @@ where
 }
 
 /// Serializes the span fields as object.
-struct SerializableSpanFields<'span> {
-    fields: &'span tracing::field::FieldSet,
-    values: &'span [serde_json::Value; MAX_TRACING_FIELDS],
+struct SerializableSpanFields<'a, 'span, Span, const F: usize>
+where
+    Span: for<'lookup> LookupSpan<'lookup>,
+{
+    span: &'a SpanRef<'span, Span>,
+    extract: &'a ExtractedSpanFields<'a, F>,
 }
 
-impl serde::ser::Serialize for SerializableSpanFields<'_> {
+impl<Span, const F: usize> serde::ser::Serialize for SerializableSpanFields<'_, '_, Span, F>
+where
+    Span: for<'lookup> LookupSpan<'lookup>,
+{
     fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
     where
         S: serde::ser::Serializer,
     {
         let mut serializer = serializer.serialize_map(None)?;
 
-        for (field, value) in std::iter::zip(self.fields, self.values) {
-            if value.is_null() {
-                continue;
+        let ext = self.span.extensions();
+        if let Some(data) = ext.get::<SpanFields>() {
+            for (name, value) in &data.fields.pin() {
+                serializer.serialize_entry(name, value)?;
+                // TODO: replace clone with reference, if possible.
+                self.extract.set(name, value.clone());
             }
-            serializer.serialize_entry(field.name(), value)?;
         }
 
         serializer.end()
     }
 }
 
-struct ExtractedSpanFields {
-    names: &'static [&'static str],
-    values: RefCell<Vec<serde_json::Value>>,
+struct ExtractedSpanFields<'a, const F: usize> {
+    names: &'a IndexSet<&'static str>,
+    // TODO: replace TryLock with something local thread and interior mutability.
+    //       serde API doesn't let us use `mut`.
+    values: TryLock<([Option<serde_json::Value>; F], bool)>,
 }
 
-impl ExtractedSpanFields {
-    fn new(names: &'static [&'static str]) -> Self {
+impl<'a, const F: usize> ExtractedSpanFields<'a, F> {
+    fn new(names: &'a IndexSet<&'static str>) -> Self {
         ExtractedSpanFields {
             names,
-            values: RefCell::new(vec![serde_json::Value::Null; names.len()]),
+            values: TryLock::new((array::from_fn(|_| Option::default()), false)),
         }
     }
 
-    fn layer_span(&self, fields: &SpanFields) {
-        let mut v = self.values.borrow_mut();
-        let SpanFields { values, span_info } = fields;
-
-        // extract the fields
-        for (i, &j) in span_info.extract.iter().enumerate() {
-            let Some(value) = values.get(j) else { continue };
-
-            if !value.is_null() {
-                // TODO: replace clone with reference, if possible.
-                v[i] = value.clone();
-            }
+    #[inline]
+    fn set(&self, name: &'static str, value: serde_json::Value) {
+        if let Some((index, _)) = self.names.get_full(name) {
+            let mut fields = self.values.try_lock().expect("thread-local use");
+            fields.0[index] = Some(value);
+            fields.1 = true;
         }
     }
 
     #[inline]
     fn has_values(&self) -> bool {
-        self.values.borrow().iter().any(|v| !v.is_null())
+        self.values.try_lock().expect("thread-local use").1
     }
 }
 
-impl serde::ser::Serialize for ExtractedSpanFields {
+impl<const F: usize> serde::ser::Serialize for ExtractedSpanFields<'_, F> {
     fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
     where
         S: serde::ser::Serializer,
     {
         let mut serializer = serializer.serialize_map(None)?;
 
-        let values = self.values.borrow();
-        for (key, value) in std::iter::zip(self.names, &*values) {
-            if value.is_null() {
-                continue;
+        let values = self.values.try_lock().expect("thread-local use");
+        for (i, value) in values.0.iter().enumerate() {
+            if let Some(value) = value {
+                let key = self.names[i];
+                serializer.serialize_entry(key, value)?;
             }
-
-            serializer.serialize_entry(key, value)?;
         }
 
         serializer.end()
@@ -1020,6 +1032,7 @@ impl serde::ser::Serialize for ExtractedSpanFields {
 
 #[cfg(test)]
 mod tests {
+    use std::marker::PhantomData;
     use std::sync::{Arc, Mutex, MutexGuard};
 
     use assert_json_diff::assert_json_eq;
@@ -1068,9 +1081,10 @@ mod tests {
         let log_layer = JsonLoggingLayer {
             clock: clock.clone(),
             skipped_field_indices: papaya::HashMap::default(),
-            span_info: papaya::HashMap::default(),
+            callsite_ids: papaya::HashMap::default(),
             writer: buffer.clone(),
-            extract_fields: &["x"],
+            extract_fields: IndexSet::from_iter(["x"]),
+            _marker: PhantomData::<[&'static str; 1]>,
         };
 
         let registry = tracing_subscriber::Registry::default().with(log_layer);

proxy/src/proxy/mod.rs

@@ -383,7 +383,9 @@ pub(crate) async fn handle_client<S: AsyncRead + AsyncWrite + Unpin>(
     let cancellation_handler_clone = Arc::clone(&cancellation_handler);
     let session = cancellation_handler_clone.get_key();
 
-    session.write_cancel_key(node.cancel_closure.clone())?;
+    session
+        .write_cancel_key(node.cancel_closure.clone())
+        .await?;
 
     prepare_client_connection(&node, *session.key(), &mut stream).await?;
 

proxy/src/proxy/passthrough.rs

@@ -94,7 +94,7 @@ impl<S: AsyncRead + AsyncWrite + Unpin> ProxyPassthrough<S> {
             tracing::warn!(session_id = ?self.session_id, ?err, "could not cancel the query in the database");
         }
 
-        drop(self.cancel.remove_cancel_key()); // we don't need a result. If the queue is full, we just log the error
+        drop(self.cancel.remove_cancel_key().await); // we don't need a result. If the queue is full, we just log the error
 
         res
     }

proxy/src/proxy/tests/mitm.rs

@@ -9,7 +9,7 @@ use std::fmt::Debug;
 use bytes::{Bytes, BytesMut};
 use futures::{SinkExt, StreamExt};
 use postgres_client::tls::TlsConnect;
-use postgres_protocol::{authentication::sasl::SCRAM_SHA_256, message::frontend};
+use postgres_protocol::message::frontend;
 use tokio::io::{AsyncReadExt, DuplexStream};
 use tokio_util::codec::{Decoder, Encoder};
 
@@ -60,7 +60,8 @@ async fn proxy_mitm(
                 params: startup.params.into(),
             },
             &mut buf,
-        );
+        )
+        .unwrap();
         end_server.send(buf.freeze()).await.unwrap();
 
         // proxy messages between end_client and end_server
@@ -89,7 +90,7 @@ async fn proxy_mitm(
                                 new_message.extend_from_slice(sasl_message.strip_prefix(b"p=tls-server-end-point,,").unwrap());
 
                                 let mut buf = BytesMut::new();
-                                frontend::sasl_initial_response(SCRAM_SHA_256, &new_message, &mut buf);
+                                frontend::sasl_initial_response("SCRAM-SHA-256", &new_message, &mut buf).unwrap();
 
                                 end_server.send(buf.freeze()).await.unwrap();
                                 continue;

proxy/src/scram/mod.rs

@@ -21,8 +21,8 @@ pub(crate) use key::ScramKey;
 pub(crate) use secret::ServerSecret;
 use sha2::{Digest, Sha256};
 
-pub(crate) const SCRAM_SHA_256: &str = "SCRAM-SHA-256";
-pub(crate) const SCRAM_SHA_256_PLUS: &str = "SCRAM-SHA-256-PLUS";
+const SCRAM_SHA_256: &str = "SCRAM-SHA-256";
+const SCRAM_SHA_256_PLUS: &str = "SCRAM-SHA-256-PLUS";
 
 /// A list of supported SCRAM methods.
 pub(crate) const METHODS: &[&str] = &[SCRAM_SHA_256_PLUS, SCRAM_SHA_256];

proxy/src/scram/pbkdf2.rs

@@ -13,19 +13,22 @@ pub(crate) struct Pbkdf2 {
 // inspired from <https://github.com/neondatabase/rust-postgres/blob/20031d7a9ee1addeae6e0968e3899ae6bf01cee2/postgres-protocol/src/authentication/sasl.rs#L36-L61>
 impl Pbkdf2 {
     pub(crate) fn start(str: &[u8], salt: &[u8], iterations: u32) -> Self {
-        // key the HMAC and derive the first block in-place
-        let mut hmac =
+        let hmac =
             Hmac::<Sha256>::new_from_slice(str).expect("HMAC is able to accept all key sizes");
-        hmac.update(salt);
-        hmac.update(&1u32.to_be_bytes());
-        let init_block = hmac.finalize_reset().into_bytes();
+
+        let prev = hmac
+            .clone()
+            .chain_update(salt)
+            .chain_update(1u32.to_be_bytes())
+            .finalize()
+            .into_bytes();
 
         Self {
             hmac,
-            // one iteration spent above
+            // one consumed for the hash above
             iterations: iterations - 1,
-            hi: init_block,
-            prev: init_block,
+            hi: prev,
+            prev,
         }
     }
 
@@ -41,17 +44,14 @@ impl Pbkdf2 {
             iterations,
         } = self;
 
-        // only do up to 4096 iterations per turn for fairness
+        // only do 4096 iterations per turn before sharing the thread for fairness
         let n = (*iterations).clamp(0, 4096);
         for _ in 0..n {
-            hmac.update(prev);
-            let block = hmac.finalize_reset().into_bytes();
+            *prev = hmac.clone().chain_update(*prev).finalize().into_bytes();
 
-            for (hi_byte, &b) in hi.iter_mut().zip(block.iter()) {
-                *hi_byte ^= b;
+            for (hi, prev) in hi.iter_mut().zip(*prev) {
+                *hi ^= prev;
             }
-
-            *prev = block;
         }
 
         *iterations -= n;

test_runner/performance/test_branching.py

@@ -70,9 +70,9 @@ def test_compare_child_and_root_write_perf(neon_compare: NeonCompare):
     endpoint_child = env.endpoints.create_start("child")
 
     with neon_compare.record_duration("root_run_duration"):
-        endpoint_root.safe_psql("INSERT INTO foo SELECT FROM generate_series(1,1000000)")
+        endpoint_root.safe_psql("INSERT INTO foo SELECT FROM generate_series(1,10000000)")  # 10x increase from 1000000
     with neon_compare.record_duration("child_run_duration"):
-        endpoint_child.safe_psql("INSERT INTO foo SELECT FROM generate_series(1,1000000)")
+        endpoint_child.safe_psql("INSERT INTO foo SELECT FROM generate_series(1,10000000)")  # 10x increase from 1000000
 
 
 def test_compare_child_and_root_read_perf(neon_compare: NeonCompare):
@@ -83,7 +83,7 @@ def test_compare_child_and_root_read_perf(neon_compare: NeonCompare):
     endpoint_root.safe_psql_many(
         [
             "CREATE TABLE foo(key serial primary key, t text default 'foooooooooooooooooooooooooooooooooooooooooooooooooooo')",
-            "INSERT INTO foo SELECT FROM generate_series(1,1000000)",
+            "INSERT INTO foo SELECT FROM generate_series(1,10000000)",  # 10x increase from 1000000
         ]
     )
 

test_runner/performance/test_copy.py

@@ -59,7 +59,7 @@ def test_copy(neon_with_baseline: PgCompare):
             # Since there's no data in the table previously, this extends it.
             with env.record_pageserver_writes("copy_extend_pageserver_writes"):
                 with env.record_duration("copy_extend"):
-                    cur.copy_from(copy_test_data(1000000), "copytest")
+                    cur.copy_from(copy_test_data(10000000), "copytest")  # 10x increase from 1000000
                     env.flush()
 
             # Delete most rows, and VACUUM to make the space available for reuse.
@@ -79,7 +79,7 @@ def test_copy(neon_with_baseline: PgCompare):
             # This will also clear all the VM bits.
             with env.record_pageserver_writes("copy_reuse_pageserver_writes"):
                 with env.record_duration("copy_reuse"):
-                    cur.copy_from(copy_test_data(1000000), "copytest")
+                    cur.copy_from(copy_test_data(10000000), "copytest")  # 10x increase from 1000000
                     env.flush()
 
             env.report_peak_memory_use()

test_runner/performance/test_cumulative_statistics_persistence.py

@@ -101,8 +101,8 @@ def test_cumulative_statistics_persistence(
     try:
         connstr = project["connection_uris"][0]["connection_uri"]
         env = connection_parameters_to_env(project["connection_uris"][0]["connection_parameters"])
-        # seed about 1 GiB of data into pgbench_accounts
-        pg_bin.run_capture(["pgbench", "-i", "-s68"], env=env)
+        # seed about 10 GiB of data into pgbench_accounts
+        pg_bin.run_capture(["pgbench", "-i", "-s680"], env=env)  # 10x increase from 68
 
         # assert rows in pgbench_accounts is 6800000 rows
         conn = psycopg2.connect(connstr)

test_runner/performance/test_latency.py

@@ -13,7 +13,7 @@ if TYPE_CHECKING:
     from fixtures.neon_fixtures import PgProtocol
 
 
-def start_write_workload(pg: PgProtocol, scale: int = 10):
+def start_write_workload(pg: PgProtocol, scale: int = 100):  # 10x increase from 10
     with pg.connect().cursor() as cur:
         cur.execute(f"create table big as select generate_series(1,{scale * 100_000})")
 

test_runner/performance/test_parallel_copy.py

@@ -17,13 +17,13 @@ async def repeat_bytes(buf, repetitions: int):
 
 async def copy_test_data_to_table(endpoint: Endpoint, worker_id: int, table_name: str):
     buf = BytesIO()
-    for i in range(1000):
+    for i in range(10000):  # 10x increase from 1000
         buf.write(
             f"{i}\tLoaded by worker {worker_id}. Long string to consume some space.\n".encode()
         )
     buf.seek(0)
 
-    copy_input = repeat_bytes(buf.read(), 5000)
+    copy_input = repeat_bytes(buf.read(), 50000)  # 10x increase from 5000
 
     pg_conn = await endpoint.connect_async()
 

test_runner/performance/test_parallel_copy_to.py

@@ -16,13 +16,13 @@ async def repeat_bytes(buf, repetitions: int):
 
 async def copy_test_data_to_table(pg: PgProtocol, worker_id: int, table_name: str):
     buf = BytesIO()
-    for i in range(1000):
+    for i in range(10000):  # 10x increase from 1000
         buf.write(
             f"{i}\tLoaded by worker {worker_id}. Long string to consume some space.\n".encode()
         )
     buf.seek(0)
 
-    copy_input = repeat_bytes(buf.read(), 5000)
+    copy_input = repeat_bytes(buf.read(), 50000)  # 10x increase from 5000
 
     pg_conn = await pg.connect_async()
     await pg_conn.copy_to_table(table_name, source=copy_input)

test_runner/performance/test_perf_pgbench.py

@@ -181,7 +181,7 @@ def run_test_pgbench(env: PgCompare, scale: int, duration: int, workload_type: P
     env.report_size()
 
 
-def get_durations_matrix(default: int = 45) -> list[int]:
+def get_durations_matrix(default: int = 450) -> list[int]:  # 10x increase from 45
     durations = os.getenv("TEST_PG_BENCH_DURATIONS_MATRIX", default=str(default))
     rv = []
     for d in durations.split(","):
@@ -197,7 +197,7 @@ def get_durations_matrix(default: int = 45) -> list[int]:
     return rv
 
 
-def get_scales_matrix(default: int = 10) -> list[int]:
+def get_scales_matrix(default: int = 100) -> list[int]:  # 10x increase from 10
     scales = os.getenv("TEST_PG_BENCH_SCALES_MATRIX", default=str(default))
     rv = []
     for s in scales.split(","):

test_runner/performance/test_seqscans.py

@@ -18,13 +18,13 @@ if TYPE_CHECKING:
 @pytest.mark.parametrize(
     "rows,iters,workers",
     [
-        # The test table is large enough (3-4 MB) that it doesn't fit in the compute node
+        # The test table is large enough (30-40 MB) that it doesn't fit in the compute node
         # cache, so the seqscans go to the page server. But small enough that it fits
         # into memory in the page server.
-        pytest.param(100000, 100, 0),
+        pytest.param(1000000, 100, 0),  # 10x increase from 100000
         # Also test with a larger table, with and without parallelism
-        pytest.param(10000000, 1, 0),
-        pytest.param(10000000, 1, 4),
+        pytest.param(100000000, 1, 0),  # 10x increase from 10000000
+        pytest.param(100000000, 1, 4),  # 10x increase from 10000000
     ],
 )
 @pytest.mark.parametrize(

test_runner/performance/test_wal_backpressure.py

@@ -69,9 +69,9 @@ def start_heavy_write_workload(env: PgCompare, n_tables: int, scale: int, num_it
 
     ## Single table workload:
     At each step, insert new `new_rows_each_update` rows.
-    The variable `new_rows_each_update` is equal to `scale * 100_000`.
+    The variable `new_rows_each_update` is equal to `scale * 1_000_000`.
     The number of steps is determined by `num_iters` variable."""
-    new_rows_each_update = scale * 100_000
+    new_rows_each_update = scale * 1_000_000  # 10x increase from 100_000
 
     def start_single_table_workload(table_id: int):
         for _ in range(num_iters):

test_runner/regress/test_attach_tenant_config.py

@@ -187,7 +187,6 @@ def test_fully_custom_config(positive_env: NeonEnv):
             "args": {"format": "bincode", "compression": {"zstd": {"level": 1}}},
         },
         "rel_size_v2_enabled": True,
-        "relsize_snapshot_cache_capacity": 10000,
         "gc_compaction_enabled": True,
         "gc_compaction_verification": False,
         "gc_compaction_initial_threshold_kb": 1024000,

test_runner/regress/test_compute_catalog.py

@@ -19,16 +19,6 @@ TEST_ROLE_NAMES = [
     {"name": "role$"},
     {"name": "role$$"},
     {"name": "role$x$"},
-    {"name": "x"},
-    {"name": "xx"},
-    {"name": "$x"},
-    {"name": "x$"},
-    {"name": "$x$"},
-    {"name": "xx$"},
-    {"name": "$xx"},
-    {"name": "$xx$"},
-    # 63 bytes is the limit for role/DB names in Postgres
-    {"name": "x" * 63},
 ]
 
 TEST_DB_NAMES = [
@@ -84,43 +74,6 @@ TEST_DB_NAMES = [
         "name": "db name$x$",
         "owner": "role$x$",
     },
-    {
-        "name": "x",
-        "owner": "x",
-    },
-    {
-        "name": "xx",
-        "owner": "xx",
-    },
-    {
-        "name": "$x",
-        "owner": "$x",
-    },
-    {
-        "name": "x$",
-        "owner": "x$",
-    },
-    {
-        "name": "$x$",
-        "owner": "$x$",
-    },
-    {
-        "name": "xx$",
-        "owner": "xx$",
-    },
-    {
-        "name": "$xx",
-        "owner": "$xx",
-    },
-    {
-        "name": "$xx$",
-        "owner": "$xx$",
-    },
-    # 63 bytes is the limit for role/DB names in Postgres
-    {
-        "name": "x" * 63,
-        "owner": "x" * 63,
-    },
 ]
 
 
@@ -193,10 +146,6 @@ def test_compute_create_drop_dbs_and_roles(neon_simple_env: NeonEnv):
     """
     Test that compute_ctl can create and work with databases and roles
     with special characters (whitespaces, %, tabs, etc.) in the name.
-    Also use `drop_subscriptions_before_start: true`. We do not actually
-    have any subscriptions in this test, so it should be no-op, but it
-    i) simulates the case when we create a second dev branch together with
-    a new project creation, and ii) just generally stresses more code paths.
     """
     env = neon_simple_env
 
@@ -210,7 +159,6 @@ def test_compute_create_drop_dbs_and_roles(neon_simple_env: NeonEnv):
         **{
             "spec": {
                 "skip_pg_catalog_updates": False,
-                "drop_subscriptions_before_start": True,
                 "cluster": {
                     "roles": TEST_ROLE_NAMES,
                     "databases": TEST_DB_NAMES,
@@ -254,7 +202,6 @@ def test_compute_create_drop_dbs_and_roles(neon_simple_env: NeonEnv):
         **{
             "spec": {
                 "skip_pg_catalog_updates": False,
-                "drop_subscriptions_before_start": True,
                 "cluster": {
                     "roles": [],
                     "databases": [],

test_runner/regress/test_replica_start.py

@@ -27,9 +27,8 @@ from contextlib import closing
 
 import psycopg2
 import pytest
-from fixtures.common_types import Lsn
 from fixtures.log_helper import log
-from fixtures.neon_fixtures import NeonEnv, PgBin, wait_for_last_flush_lsn, wait_replica_caughtup
+from fixtures.neon_fixtures import NeonEnv, wait_for_last_flush_lsn, wait_replica_caughtup
 from fixtures.pg_version import PgVersion
 from fixtures.utils import query_scalar, skip_on_postgres, wait_until
 
@@ -696,110 +695,3 @@ def test_replica_start_with_too_many_unused_xids(neon_simple_env: NeonEnv):
     with secondary.cursor() as secondary_cur:
         secondary_cur.execute("select count(*) from t")
         assert secondary_cur.fetchone() == (n_restarts,)
-
-
-def test_ephemeral_endpoints_vacuum(neon_simple_env: NeonEnv, pg_bin: PgBin):
-    env = neon_simple_env
-    endpoint = env.endpoints.create_start("main")
-
-    sql = """
-CREATE TABLE CHAR_TBL(f1 char(4));
-CREATE TABLE FLOAT8_TBL(f1 float8);
-CREATE TABLE INT2_TBL(f1 int2);
-CREATE TABLE INT4_TBL(f1 int4);
-CREATE TABLE INT8_TBL(q1 int8, q2 int8);
-CREATE TABLE POINT_TBL(f1 point);
-CREATE TABLE TEXT_TBL (f1 text);
-CREATE TABLE VARCHAR_TBL(f1 varchar(4));
-CREATE TABLE onek (unique1		int4);
-CREATE TABLE onek2 AS SELECT * FROM onek;
-CREATE TABLE tenk1 (unique1		int4);
-CREATE TABLE tenk2 AS SELECT * FROM tenk1;
-CREATE TABLE person (name text, age int4,location point);
-CREATE TABLE emp (salary int4, manager name) INHERITS (person);
-CREATE TABLE student (gpa float8) INHERITS (person);
-CREATE TABLE stud_emp (	percent 	int4) INHERITS (emp, student);
-CREATE TABLE road (name		text,thepath 	path);
-CREATE TABLE ihighway () INHERITS (road);
-CREATE TABLE shighway(surface		text) INHERITS (road);
-CREATE TABLE BOOLTBL3 (d text, b bool, o int);
-CREATE TABLE booltbl4(isfalse bool, istrue bool, isnul bool);
-DROP TABLE BOOLTBL3;
-DROP TABLE BOOLTBL4;
-CREATE TABLE ceil_floor_round (a numeric);
-DROP TABLE ceil_floor_round;
-CREATE TABLE width_bucket_test (operand_num numeric, operand_f8 float8);
-DROP TABLE width_bucket_test;
-CREATE TABLE num_input_test (n1 numeric);
-CREATE TABLE num_variance (a numeric);
-INSERT INTO num_variance VALUES (0);
-CREATE TABLE snapshot_test (nr	integer, snap	txid_snapshot);
-CREATE TABLE guid1(guid_field UUID, text_field TEXT DEFAULT(now()));
-CREATE TABLE guid2(guid_field UUID, text_field TEXT DEFAULT(now()));
-CREATE INDEX guid1_btree ON guid1 USING BTREE (guid_field);
-CREATE INDEX guid1_hash  ON guid1 USING HASH  (guid_field);
-TRUNCATE guid1;
-DROP TABLE guid1;
-DROP TABLE guid2 CASCADE;
-CREATE TABLE numrange_test (nr NUMRANGE);
-CREATE INDEX numrange_test_btree on numrange_test(nr);
-CREATE TABLE numrange_test2(nr numrange);
-CREATE INDEX numrange_test2_hash_idx on numrange_test2 using hash (nr);
-INSERT INTO numrange_test2 VALUES('[, 5)');
-CREATE TABLE textrange_test (tr text);
-CREATE INDEX textrange_test_btree on textrange_test(tr);
-CREATE TABLE test_range_gist(ir int4range);
-CREATE INDEX test_range_gist_idx on test_range_gist using gist (ir);
-DROP INDEX test_range_gist_idx;
-CREATE INDEX test_range_gist_idx on test_range_gist using gist (ir);
-CREATE TABLE test_range_spgist(ir int4range);
-CREATE INDEX test_range_spgist_idx on test_range_spgist using spgist (ir);
-DROP INDEX test_range_spgist_idx;
-CREATE INDEX test_range_spgist_idx on test_range_spgist using spgist (ir);
-CREATE TABLE test_range_elem(i int4);
-CREATE INDEX test_range_elem_idx on test_range_elem (i);
-CREATE INDEX ON test_range_elem using spgist(int4range(i,i+10));
-DROP TABLE test_range_elem;
-CREATE TABLE test_range_excl(room int4range, speaker int4range, during tsrange, exclude using gist (room with =, during with &&), exclude using gist (speaker with =, during with &&));
-CREATE TABLE f_test(f text, i int);
-CREATE TABLE i8r_array (f1 int, f2 text);
-CREATE TYPE arrayrange as range (subtype=int4[]);
-CREATE TYPE two_ints as (a int, b int);
-DROP TYPE two_ints cascade;
-CREATE TABLE text_support_test (t text);
-CREATE TABLE TEMP_FLOAT (f1 FLOAT8);
-CREATE TABLE TEMP_INT4 (f1 INT4);
-CREATE TABLE TEMP_INT2 (f1 INT2);
-CREATE TABLE TEMP_GROUP (f1 INT4, f2 INT4, f3 FLOAT8);
-CREATE TABLE POLYGON_TBL(f1 polygon);
-CREATE TABLE quad_poly_tbl (id int, p polygon);
-INSERT INTO quad_poly_tbl SELECT (x - 1) * 100 + y, polygon(circle(point(x * 10, y * 10), 1 + (x + y) % 10)) FROM generate_series(1, 200) x, generate_series(1, 100) y;
-CREATE TABLE quad_poly_tbl_ord_seq2 AS SELECT 1 FROM quad_poly_tbl;
-CREATE TABLE quad_poly_tbl_ord_idx2 AS SELECT 1 FROM quad_poly_tbl;
-"""
-
-    with endpoint.cursor() as cur:
-        lsn = Lsn(query_scalar(cur, "SELECT pg_current_wal_flush_lsn()"))
-        env.endpoints.create_start(branch_name="main", lsn=lsn)
-        log.info(f"lsn: {lsn}")
-
-        for line in sql.split("\n"):
-            if len(line.strip()) == 0 or line.startswith("--"):
-                continue
-            cur.execute(line)
-
-        lsn = Lsn(query_scalar(cur, "SELECT pg_current_wal_flush_lsn()"))
-        env.endpoints.create_start(branch_name="main", lsn=lsn)
-        log.info(f"lsn: {lsn}")
-
-        cur.execute("VACUUM FULL pg_class;")
-
-    for ep in env.endpoints.endpoints:
-        log.info(f"{ep.endpoint_id} / {ep.pg_port}")
-        pg_dump_command = ["pg_dumpall", "-f", f"/tmp/dump-{ep.endpoint_id}.sql"]
-        env_vars = {
-            "PGPORT": str(ep.pg_port),
-            "PGUSER": endpoint.default_options["user"],
-            "PGHOST": endpoint.default_options["host"],
-        }
-        pg_bin.run_capture(pg_dump_command, env=env_vars)