chore(compute): Do a minor cleanup of remote extensions code

## Problem

We were logging a warning after a single request timeout, while listing
objects.

Closes: https://github.com/neondatabase/neon/issues/10166

## Summary of changes

- These timeouts are a pretty normal part of life, so back it off to
only log a warning after two in a row.

.github/workflows/build_and_test.yml

@@ -820,8 +820,8 @@ jobs:
       - name: Print logs and clean up
         if: always()
         run: |
-          docker compose -f ./docker-compose/docker-compose.yml logs || 0
-          docker compose -f ./docker-compose/docker-compose.yml down
+          docker compose --profile test-extensions -f ./docker-compose/docker-compose.yml logs || true
+          docker compose --profile test-extensions -f ./docker-compose/docker-compose.yml down
 
   promote-images-dev:
     needs: [ check-permissions, tag, vm-compute-node-image, neon-image ]
@@ -859,7 +859,7 @@ jobs:
           done
 
   promote-images-prod:
-    needs: [ check-permissions, tag, test-images, vm-compute-node-image ]
+    needs: [ check-permissions, tag, test-images, promote-images-dev ]
     runs-on: ubuntu-22.04
     if: github.ref_name == 'main' || github.ref_name == 'release' || github.ref_name == 'release-proxy' || github.ref_name == 'release-compute'
 

.github/workflows/release.yml

@@ -3,8 +3,9 @@ name: Create Release Branch
 on:
   schedule:
     # It should be kept in sync with if-condition in jobs
-    - cron: '0 6 * * FRI' # Storage release
     - cron: '0 6 * * THU' # Proxy release
+    - cron: '0 6 * * FRI' # Storage release
+    - cron: '0 7 * * FRI' # Compute release
   workflow_dispatch:
     inputs:
       create-storage-release-branch:
@@ -55,7 +56,7 @@ jobs:
       ci-access-token: ${{ secrets.CI_ACCESS_TOKEN }}
 
   create-compute-release-branch:
-    if: inputs.create-compute-release-branch
+    if: ${{ github.event.schedule == '0 7 * * FRI' || inputs.create-compute-release-branch }}
 
     permissions:
       contents: write

Cargo.lock

@@ -942,6 +942,18 @@ version = "1.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8c3c1a368f70d6cf7302d78f8f7093da241fb8e8807c05cc9e51a125895a6d5b"
 
+[[package]]
+name = "bb8"
+version = "0.8.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d89aabfae550a5c44b43ab941844ffcd2e993cb6900b342debf59e9ea74acdb8"
+dependencies = [
+ "async-trait",
+ "futures-util",
+ "parking_lot 0.12.1",
+ "tokio",
+]
+
 [[package]]
 name = "bcder"
 version = "0.7.4"
@@ -1301,7 +1313,7 @@ dependencies = [
  "tar",
  "thiserror",
  "tokio",
- "tokio-postgres",
+ "tokio-postgres 0.7.7",
  "tokio-stream",
  "tokio-util",
  "tower 0.5.2",
@@ -1410,7 +1422,7 @@ dependencies = [
  "storage_broker",
  "thiserror",
  "tokio",
- "tokio-postgres",
+ "tokio-postgres 0.7.7",
  "tokio-util",
  "toml",
  "toml_edit",
@@ -1786,11 +1798,24 @@ dependencies = [
  "chrono",
  "diesel_derives",
  "itoa",
- "pq-sys",
- "r2d2",
  "serde_json",
 ]
 
+[[package]]
+name = "diesel-async"
+version = "0.5.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "51a307ac00f7c23f526a04a77761a0519b9f0eb2838ebf5b905a58580095bdcb"
+dependencies = [
+ "async-trait",
+ "bb8",
+ "diesel",
+ "futures-util",
+ "scoped-futures",
+ "tokio",
+ "tokio-postgres 0.7.12",
+]
+
 [[package]]
 name = "diesel_derives"
 version = "2.2.1"
@@ -4042,8 +4067,8 @@ dependencies = [
  "pageserver_compaction",
  "pin-project-lite",
  "postgres",
- "postgres-protocol",
- "postgres-types",
+ "postgres-protocol 0.6.4",
+ "postgres-types 0.2.4",
  "postgres_backend",
  "postgres_connection",
  "postgres_ffi",
@@ -4074,7 +4099,7 @@ dependencies = [
  "tokio",
  "tokio-epoll-uring",
  "tokio-io-timeout",
- "tokio-postgres",
+ "tokio-postgres 0.7.7",
  "tokio-stream",
  "tokio-tar",
  "tokio-util",
@@ -4132,7 +4157,7 @@ dependencies = [
  "serde",
  "thiserror",
  "tokio",
- "tokio-postgres",
+ "tokio-postgres 0.7.7",
  "tokio-stream",
  "tokio-util",
  "utils",
@@ -4438,7 +4463,7 @@ dependencies = [
  "futures-util",
  "log",
  "tokio",
- "tokio-postgres",
+ "tokio-postgres 0.7.7",
 ]
 
 [[package]]
@@ -4459,6 +4484,24 @@ dependencies = [
  "stringprep",
 ]
 
+[[package]]
+name = "postgres-protocol"
+version = "0.6.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "acda0ebdebc28befa84bee35e651e4c5f09073d668c7aed4cf7e23c3cda84b23"
+dependencies = [
+ "base64 0.22.1",
+ "byteorder",
+ "bytes",
+ "fallible-iterator",
+ "hmac",
+ "md-5",
+ "memchr",
+ "rand 0.8.5",
+ "sha2",
+ "stringprep",
+]
+
 [[package]]
 name = "postgres-protocol2"
 version = "0.1.0"
@@ -4482,7 +4525,18 @@ source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#511f
 dependencies = [
  "bytes",
  "fallible-iterator",
- "postgres-protocol",
+ "postgres-protocol 0.6.4",
+]
+
+[[package]]
+name = "postgres-types"
+version = "0.2.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f66ea23a2d0e5734297357705193335e0a957696f34bed2f2faefacb2fec336f"
+dependencies = [
+ "bytes",
+ "fallible-iterator",
+ "postgres-protocol 0.6.7",
 ]
 
 [[package]]
@@ -4507,7 +4561,7 @@ dependencies = [
  "serde",
  "thiserror",
  "tokio",
- "tokio-postgres",
+ "tokio-postgres 0.7.7",
  "tokio-postgres-rustls",
  "tokio-rustls 0.26.0",
  "tokio-util",
@@ -4522,7 +4576,7 @@ dependencies = [
  "itertools 0.10.5",
  "once_cell",
  "postgres",
- "tokio-postgres",
+ "tokio-postgres 0.7.7",
  "url",
 ]
 
@@ -4609,15 +4663,6 @@ version = "0.2.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5b40af805b3121feab8a3c29f04d8ad262fa8e0561883e7653e024ae4479e6de"
 
-[[package]]
-name = "pq-sys"
-version = "0.6.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f6cc05d7ea95200187117196eee9edd0644424911821aeb28a18ce60ea0b8793"
-dependencies = [
- "vcpkg",
-]
-
 [[package]]
 name = "pq_proto"
 version = "0.1.0"
@@ -4625,7 +4670,7 @@ dependencies = [
  "byteorder",
  "bytes",
  "itertools 0.10.5",
- "postgres-protocol",
+ "postgres-protocol 0.6.4",
  "rand 0.8.5",
  "serde",
  "thiserror",
@@ -4873,7 +4918,7 @@ dependencies = [
  "tikv-jemalloc-ctl",
  "tikv-jemallocator",
  "tokio",
- "tokio-postgres",
+ "tokio-postgres 0.7.7",
  "tokio-postgres2",
  "tokio-rustls 0.26.0",
  "tokio-tungstenite 0.21.0",
@@ -4930,17 +4975,6 @@ dependencies = [
  "proc-macro2",
 ]
 
-[[package]]
-name = "r2d2"
-version = "0.8.10"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "51de85fb3fb6524929c8a2eb85e6b6d363de4e8c48f9e2c2eac4944abc181c93"
-dependencies = [
- "log",
- "parking_lot 0.12.1",
- "scheduled-thread-pool",
-]
-
 [[package]]
 name = "rand"
 version = "0.7.3"
@@ -5672,7 +5706,7 @@ dependencies = [
  "pageserver_api",
  "parking_lot 0.12.1",
  "postgres",
- "postgres-protocol",
+ "postgres-protocol 0.6.4",
  "postgres_backend",
  "postgres_ffi",
  "pprof",
@@ -5696,7 +5730,7 @@ dependencies = [
  "tikv-jemallocator",
  "tokio",
  "tokio-io-timeout",
- "tokio-postgres",
+ "tokio-postgres 0.7.7",
  "tokio-stream",
  "tokio-tar",
  "tokio-util",
@@ -5755,12 +5789,12 @@ dependencies = [
 ]
 
 [[package]]
-name = "scheduled-thread-pool"
-version = "0.2.7"
+name = "scoped-futures"
+version = "0.1.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3cbc66816425a074528352f5789333ecff06ca41b36b0b0efdfbb29edc391a19"
+checksum = "1b24aae2d0636530f359e9d5ef0c04669d11c5e756699b27a6a6d845d8329091"
 dependencies = [
- "parking_lot 0.12.1",
+ "pin-project-lite",
 ]
 
 [[package]]
@@ -6295,6 +6329,7 @@ dependencies = [
  "clap",
  "control_plane",
  "diesel",
+ "diesel-async",
  "diesel_migrations",
  "fail",
  "futures",
@@ -6309,10 +6344,10 @@ dependencies = [
  "pageserver_api",
  "pageserver_client",
  "postgres_connection",
- "r2d2",
  "rand 0.8.5",
  "reqwest",
  "routerify",
+ "scoped-futures",
  "scopeguard",
  "serde",
  "serde_json",
@@ -6365,7 +6400,7 @@ dependencies = [
  "serde_json",
  "storage_controller_client",
  "tokio",
- "tokio-postgres",
+ "tokio-postgres 0.7.7",
  "tokio-postgres-rustls",
  "tokio-stream",
  "tokio-util",
@@ -6774,7 +6809,7 @@ dependencies = [
 [[package]]
 name = "tokio-epoll-uring"
 version = "0.1.0"
-source = "git+https://github.com/neondatabase/tokio-epoll-uring.git?branch=main#33e00106a268644d02ba0461bbd64476073b0ee1"
+source = "git+https://github.com/neondatabase/tokio-epoll-uring.git?branch=main#781989bb540a1408b0b93daa1e9d1fa452195497"
 dependencies = [
  "futures",
  "nix 0.26.4",
@@ -6824,13 +6859,39 @@ dependencies = [
  "percent-encoding",
  "phf",
  "pin-project-lite",
- "postgres-protocol",
- "postgres-types",
+ "postgres-protocol 0.6.4",
+ "postgres-types 0.2.4",
  "socket2",
  "tokio",
  "tokio-util",
 ]
 
+[[package]]
+name = "tokio-postgres"
+version = "0.7.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3b5d3742945bc7d7f210693b0c58ae542c6fd47b17adbbda0885f3dcb34a6bdb"
+dependencies = [
+ "async-trait",
+ "byteorder",
+ "bytes",
+ "fallible-iterator",
+ "futures-channel",
+ "futures-util",
+ "log",
+ "parking_lot 0.12.1",
+ "percent-encoding",
+ "phf",
+ "pin-project-lite",
+ "postgres-protocol 0.6.7",
+ "postgres-types 0.2.8",
+ "rand 0.8.5",
+ "socket2",
+ "tokio",
+ "tokio-util",
+ "whoami",
+]
+
 [[package]]
 name = "tokio-postgres-rustls"
 version = "0.12.0"
@@ -6840,7 +6901,7 @@ dependencies = [
  "ring",
  "rustls 0.23.18",
  "tokio",
- "tokio-postgres",
+ "tokio-postgres 0.7.7",
  "tokio-rustls 0.26.0",
  "x509-certificate",
 ]
@@ -7369,7 +7430,7 @@ dependencies = [
 [[package]]
 name = "uring-common"
 version = "0.1.0"
-source = "git+https://github.com/neondatabase/tokio-epoll-uring.git?branch=main#33e00106a268644d02ba0461bbd64476073b0ee1"
+source = "git+https://github.com/neondatabase/tokio-epoll-uring.git?branch=main#781989bb540a1408b0b93daa1e9d1fa452195497"
 dependencies = [
  "bytes",
  "io-uring",
@@ -7498,12 +7559,6 @@ version = "0.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "830b7e5d4d90034032940e4ace0d9a9a057e7a45cd94e6c007832e39edb82f6d"
 
-[[package]]
-name = "vcpkg"
-version = "0.2.15"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426"
-
 [[package]]
 name = "version_check"
 version = "0.9.4"
@@ -7523,7 +7578,7 @@ dependencies = [
  "serde_json",
  "sysinfo",
  "tokio",
- "tokio-postgres",
+ "tokio-postgres 0.7.7",
  "tokio-util",
  "tracing",
  "tracing-subscriber",

Makefile

@@ -64,8 +64,6 @@ CARGO_BUILD_FLAGS += $(filter -j1,$(MAKEFLAGS))
 CARGO_CMD_PREFIX += $(if $(filter n,$(MAKEFLAGS)),,+)
 # Force cargo not to print progress bar
 CARGO_CMD_PREFIX += CARGO_TERM_PROGRESS_WHEN=never CI=1
-# Set PQ_LIB_DIR to make sure `storage_controller` get linked with bundled libpq (through diesel)
-CARGO_CMD_PREFIX += PQ_LIB_DIR=$(POSTGRES_INSTALL_DIR)/v16/lib
 
 CACHEDIR_TAG_CONTENTS := "Signature: 8a477f597d28d172789f06886806bc55"
 

README.md

@@ -21,8 +21,10 @@ The Neon storage engine consists of two major components:
 
 See developer documentation in [SUMMARY.md](/docs/SUMMARY.md) for more information.
 
-## Running local installation
+## Running a local development environment
 
+Neon can be run on a workstation for small experiments and to test code changes, by
+following these instructions.
 
 #### Installing dependencies on Linux
 1. Install build dependencies and other applicable packages
@@ -238,7 +240,7 @@ postgres=# select * from t;
 > cargo neon stop
 ```
 
-More advanced usages can be found at [Control Plane and Neon Local](./control_plane/README.md).
+More advanced usages can be found at [Local Development Control Plane (`neon_local`))](./control_plane/README.md).
 
 #### Handling build failures
 

compute/compute-node.Dockerfile

@@ -67,6 +67,9 @@ RUN cd postgres && \
     # Enable some of contrib extensions
     echo 'trusted = true' >> /usr/local/pgsql/share/extension/autoinc.control && \
     echo 'trusted = true' >> /usr/local/pgsql/share/extension/dblink.control && \
+    echo 'trusted = true' >> /usr/local/pgsql/share/extension/postgres_fdw.control && \
+    file=/usr/local/pgsql/share/extension/postgres_fdw--1.0.sql && [ -e $file ] && \
+    echo 'GRANT USAGE ON FOREIGN DATA WRAPPER postgres_fdw TO neon_superuser;' >> $file && \
     echo 'trusted = true' >> /usr/local/pgsql/share/extension/bloom.control && \
     echo 'trusted = true' >> /usr/local/pgsql/share/extension/earthdistance.control && \
     echo 'trusted = true' >> /usr/local/pgsql/share/extension/insert_username.control && \
@@ -1263,11 +1266,12 @@ RUN set -e \
 
 #########################################################################################
 #
-# Layers "postgres-exporter" and "sql-exporter"
+# Layers "postgres-exporter", "pgbouncer-exporter", and "sql-exporter"
 #
 #########################################################################################
 
 FROM quay.io/prometheuscommunity/postgres-exporter:v0.16.0 AS postgres-exporter
+FROM quay.io/prometheuscommunity/pgbouncer-exporter:v0.10.2 AS pgbouncer-exporter
 
 # Keep the version the same as in build-tools.Dockerfile and
 # test_runner/regress/test_compute_metrics.py.
@@ -1399,6 +1403,7 @@ RUN mkdir -p /etc/local_proxy && chown postgres:postgres /etc/local_proxy
 
 # Metrics exporter binaries and  configuration files
 COPY --from=postgres-exporter /bin/postgres_exporter /bin/postgres_exporter
+COPY --from=pgbouncer-exporter /bin/pgbouncer_exporter /bin/pgbouncer_exporter
 COPY --from=sql-exporter      /bin/sql_exporter      /bin/sql_exporter
 
 COPY --chown=postgres compute/etc/postgres_exporter.yml /etc/postgres_exporter.yml

compute/etc/pgbouncer.ini

@@ -19,6 +19,8 @@ max_prepared_statements=0
 admin_users=postgres
 unix_socket_dir=/tmp/
 unix_socket_mode=0777
+; required for pgbouncer_exporter
+ignore_startup_parameters=extra_float_digits
 
 ;; Disable connection logging. It produces a lot of logs that no one looks at,
 ;; and we can get similar log entries from the proxy too. We had incidents in

compute/vm-image-spec-bookworm.yaml

@@ -27,6 +27,10 @@ commands:
     user: nobody
     sysvInitAction: respawn
     shell: 'DATA_SOURCE_NAME="user=cloud_admin sslmode=disable dbname=postgres application_name=postgres-exporter" /bin/postgres_exporter --config.file=/etc/postgres_exporter.yml'
+  - name: pgbouncer-exporter
+    user: postgres
+    sysvInitAction: respawn
+    shell: '/bin/pgbouncer_exporter --pgBouncer.connectionString="postgres:///pgbouncer?host=/tmp&port=6432&dbname=pgbouncer&user=pgbouncer"'
   - name: sql-exporter
     user: nobody
     sysvInitAction: respawn

compute/vm-image-spec-bullseye.yaml

@@ -27,6 +27,10 @@ commands:
     user: nobody
     sysvInitAction: respawn
     shell: 'DATA_SOURCE_NAME="user=cloud_admin sslmode=disable dbname=postgres application_name=postgres-exporter" /bin/postgres_exporter --config.file=/etc/postgres_exporter.yml'
+  - name: pgbouncer-exporter
+    user: postgres
+    sysvInitAction: respawn
+    shell: '/bin/pgbouncer_exporter --pgBouncer.connectionString="postgres:///pgbouncer?host=/tmp&port=6432&dbname=pgbouncer&user=pgbouncer"'
   - name: sql-exporter
     user: nobody
     sysvInitAction: respawn

compute_tools/src/bin/fast_import.rs

@@ -31,7 +31,6 @@ use camino::{Utf8Path, Utf8PathBuf};
 use clap::Parser;
 use compute_tools::extension_server::{get_pg_version, PostgresMajorVersion};
 use nix::unistd::Pid;
-use tokio::io::BufReader;
 use tracing::{error, info, info_span, warn, Instrument};
 use utils::fs_ext::is_directory_empty;
 
@@ -53,8 +52,6 @@ struct Args {
     s3_prefix: Option<s3_uri::S3Uri>,
     #[clap(long)]
     source_connection_string: Option<String>,
-    #[clap(long)]
-    restore_connection_string: Option<String>, // will not run postgres if specified, will do pg_restore to this connection string
     #[clap(short, long)]
     interactive: bool,
     #[clap(long)]
@@ -71,8 +68,6 @@ struct Spec {
     encryption_secret: EncryptionSecret,
     #[serde_as(as = "serde_with::base64::Base64")]
     source_connstring_ciphertext_base64: Vec<u8>,
-    #[serde_as(as = "Option<serde_with::base64::Base64>")]
-    restore_connstring_ciphertext_base64: Option<Vec<u8>>,
 }
 
 #[derive(serde::Deserialize)]
@@ -88,189 +83,6 @@ const DEFAULT_LOCALE: &str = if cfg!(target_os = "macos") {
     "C.UTF-8"
 };
 
-async fn decode_connstring(
-    kms_client: &aws_sdk_kms::Client,
-    key_id: &String,
-    connstring_ciphertext_base64: Vec<u8>,
-) -> Result<String, anyhow::Error> {
-    let mut output = kms_client
-        .decrypt()
-        .key_id(key_id)
-        .ciphertext_blob(aws_sdk_s3::primitives::Blob::new(
-            connstring_ciphertext_base64,
-        ))
-        .send()
-        .await
-        .context("decrypt connection string")?;
-
-    let plaintext = output
-        .plaintext
-        .take()
-        .context("get plaintext connection string")?;
-
-    String::from_utf8(plaintext.into_inner()).context("parse connection string as utf8")
-}
-
-struct PostgresProcess {
-    pgdata_dir: Utf8PathBuf,
-    pg_bin_dir: Utf8PathBuf,
-    pgbin: Utf8PathBuf,
-    pg_lib_dir: Utf8PathBuf,
-    postgres_proc: Option<tokio::process::Child>,
-}
-
-impl PostgresProcess {
-    fn new(pgdata_dir: Utf8PathBuf, pg_bin_dir: Utf8PathBuf, pg_lib_dir: Utf8PathBuf) -> Self {
-        Self {
-            pgdata_dir,
-            pgbin: pg_bin_dir.join("postgres"),
-            pg_bin_dir,
-            pg_lib_dir,
-            postgres_proc: None,
-        }
-    }
-
-    async fn prepare(&self, initdb_user: &str) -> Result<(), anyhow::Error> {
-        tokio::fs::create_dir(&self.pgdata_dir)
-            .await
-            .context("create pgdata directory")?;
-
-        let pg_version = match get_pg_version(self.pgbin.as_ref()) {
-            PostgresMajorVersion::V14 => 14,
-            PostgresMajorVersion::V15 => 15,
-            PostgresMajorVersion::V16 => 16,
-            PostgresMajorVersion::V17 => 17,
-        };
-        postgres_initdb::do_run_initdb(postgres_initdb::RunInitdbArgs {
-            superuser: initdb_user,
-            locale: DEFAULT_LOCALE, // XXX: this shouldn't be hard-coded,
-            pg_version,
-            initdb_bin: self.pg_bin_dir.join("initdb").as_ref(),
-            library_search_path: &self.pg_lib_dir, // TODO: is this right? Prob works in compute image, not sure about neon_local.
-            pgdata: &self.pgdata_dir,
-        })
-        .await
-        .context("initdb")
-    }
-
-    async fn start(
-        &mut self,
-        initdb_user: &str,
-        port: u16,
-        nproc: usize,
-    ) -> Result<&tokio::process::Child, anyhow::Error> {
-        self.prepare(initdb_user).await?;
-
-        //
-        // Launch postgres process
-        //
-        let mut proc = tokio::process::Command::new(&self.pgbin)
-            .arg("-D")
-            .arg(&self.pgdata_dir)
-            .args(["-p", &format!("{port}")])
-            .args(["-c", "wal_level=minimal"])
-            .args(["-c", "shared_buffers=10GB"])
-            .args(["-c", "max_wal_senders=0"])
-            .args(["-c", "fsync=off"])
-            .args(["-c", "full_page_writes=off"])
-            .args(["-c", "synchronous_commit=off"])
-            .args(["-c", "maintenance_work_mem=8388608"])
-            .args(["-c", &format!("max_parallel_maintenance_workers={nproc}")])
-            .args(["-c", &format!("max_parallel_workers={nproc}")])
-            .args(["-c", &format!("max_parallel_workers_per_gather={nproc}")])
-            .args(["-c", &format!("max_worker_processes={nproc}")])
-            .args(["-c", "effective_io_concurrency=100"])
-            .env_clear()
-            .env("LD_LIBRARY_PATH", &self.pg_lib_dir)
-            .stdout(std::process::Stdio::piped())
-            .stderr(std::process::Stdio::piped())
-            .spawn()
-            .context("spawn postgres")?;
-
-        info!("spawned postgres, waiting for it to become ready");
-        tokio::spawn(
-            child_stdio_to_log::relay_process_output(proc.stdout.take(), proc.stderr.take())
-                .instrument(info_span!("postgres")),
-        );
-
-        self.postgres_proc = Some(proc);
-        Ok(self.postgres_proc.as_ref().unwrap())
-    }
-
-    async fn shutdown(&mut self) -> Result<(), anyhow::Error> {
-        let proc: &mut tokio::process::Child = self.postgres_proc.as_mut().unwrap();
-        info!("shutdown postgres");
-        {
-            nix::sys::signal::kill(
-                Pid::from_raw(i32::try_from(proc.id().unwrap()).expect("convert child pid to i32")),
-                nix::sys::signal::SIGTERM,
-            )
-            .context("signal postgres to shut down")?;
-            proc.wait()
-                .await
-                .context("wait for postgres to shut down")?;
-        }
-        Ok(())
-    }
-}
-
-async fn wait_until_ready(connstring: String, create_dbname: String) {
-    // Create neondb database in the running postgres
-    let start_time = std::time::Instant::now();
-
-    loop {
-        if start_time.elapsed() > PG_WAIT_TIMEOUT {
-            error!(
-                "timeout exceeded: failed to poll postgres and create database within 10 minutes"
-            );
-            std::process::exit(1);
-        }
-
-        match tokio_postgres::connect(
-            &connstring.replace("dbname=neondb", "dbname=postgres"),
-            tokio_postgres::NoTls,
-        )
-        .await
-        {
-            Ok((client, connection)) => {
-                // Spawn the connection handling task to maintain the connection
-                tokio::spawn(async move {
-                    if let Err(e) = connection.await {
-                        warn!("connection error: {}", e);
-                    }
-                });
-
-                match client
-                    .simple_query(format!("CREATE DATABASE {create_dbname};").as_str())
-                    .await
-                {
-                    Ok(_) => {
-                        info!("created {} database", create_dbname);
-                        break;
-                    }
-                    Err(e) => {
-                        warn!(
-                            "failed to create database: {}, retying in {}s",
-                            e,
-                            PG_WAIT_RETRY_INTERVAL.as_secs_f32()
-                        );
-                        tokio::time::sleep(PG_WAIT_RETRY_INTERVAL).await;
-                        continue;
-                    }
-                }
-            }
-            Err(_) => {
-                info!(
-                    "postgres not ready yet, retrying in {}s",
-                    PG_WAIT_RETRY_INTERVAL.as_secs_f32()
-                );
-                tokio::time::sleep(PG_WAIT_RETRY_INTERVAL).await;
-                continue;
-            }
-        }
-    }
-}
-
 #[tokio::main]
 pub(crate) async fn main() -> anyhow::Result<()> {
     utils::logging::init(
@@ -294,6 +106,10 @@ pub(crate) async fn main() -> anyhow::Result<()> {
     let working_directory = args.working_directory;
     let pg_bin_dir = args.pg_bin_dir;
     let pg_lib_dir = args.pg_lib_dir;
+    let pg_port = args.pg_port.unwrap_or_else(|| {
+        info!("pg_port not specified, using default 5432");
+        5432
+    });
 
     // Initialize AWS clients only if s3_prefix is specified
     let (aws_config, kms_client) = if args.s3_prefix.is_some() {
@@ -304,18 +120,8 @@ pub(crate) async fn main() -> anyhow::Result<()> {
         (None, None)
     };
 
-    let superuser = "cloud_admin";
-    let pg_port = || {
-        args.pg_port.unwrap_or_else(|| {
-            info!("pg_port not specified, using default 5432");
-            5432
-        })
-    };
-
-    let mut run_postgres = true;
-
-    // Get connection strings either from S3 spec or direct arguments
-    let (source_connstring, restore_connstring) = if let Some(s3_prefix) = &args.s3_prefix {
+    // Get source connection string either from S3 spec or direct argument
+    let source_connection_string = if let Some(s3_prefix) = &args.s3_prefix {
         let spec: Spec = {
             let spec_key = s3_prefix.append("/spec.json");
             let s3_client = aws_sdk_s3::Client::new(aws_config.as_ref().unwrap());
@@ -335,47 +141,28 @@ pub(crate) async fn main() -> anyhow::Result<()> {
 
         match spec.encryption_secret {
             EncryptionSecret::KMS { key_id } => {
-                let source = decode_connstring(
-                    kms_client.as_ref().unwrap(),
-                    &key_id,
-                    spec.source_connstring_ciphertext_base64,
-                )
-                .await?;
-
-                let restore =
-                    if let Some(restore_ciphertext) = spec.restore_connstring_ciphertext_base64 {
-                        run_postgres = false;
-                        decode_connstring(kms_client.as_ref().unwrap(), &key_id, restore_ciphertext)
-                            .await?
-                    } else {
-                        // restoring to local postgres otherwise
-                        format!(
-                            "host=localhost port={} user={} dbname=neondb",
-                            pg_port(),
-                            superuser
-                        )
-                    };
-
-                (source, restore)
+                let mut output = kms_client
+                    .unwrap()
+                    .decrypt()
+                    .key_id(key_id)
+                    .ciphertext_blob(aws_sdk_s3::primitives::Blob::new(
+                        spec.source_connstring_ciphertext_base64,
+                    ))
+                    .send()
+                    .await
+                    .context("decrypt source connection string")?;
+                let plaintext = output
+                    .plaintext
+                    .take()
+                    .context("get plaintext source connection string")?;
+                String::from_utf8(plaintext.into_inner())
+                    .context("parse source connection string as utf8")?
             }
         }
     } else {
-        (
-            args.source_connection_string.unwrap(),
-            if let Some(val) = args.restore_connection_string {
-                run_postgres = false;
-                val
-            } else {
-                format!(
-                    "host=localhost port={} user={} dbname=neondb",
-                    pg_port(),
-                    superuser
-                )
-            },
-        )
+        args.source_connection_string.unwrap()
     };
 
-    // unused if run_postgres is false, but needed for shutdown
     match tokio::fs::create_dir(&working_directory).await {
         Ok(()) => {}
         Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
@@ -390,21 +177,127 @@ pub(crate) async fn main() -> anyhow::Result<()> {
         }
         Err(e) => return Err(anyhow::Error::new(e).context("create working directory")),
     }
+
     let pgdata_dir = working_directory.join("pgdata");
+    tokio::fs::create_dir(&pgdata_dir)
+        .await
+        .context("create pgdata directory")?;
 
-    let postgres_proc = if run_postgres {
-        assert!(restore_connstring.contains("host=localhost"));
-        let mut proc =
-            PostgresProcess::new(pgdata_dir.clone(), pg_bin_dir.clone(), pg_lib_dir.clone());
-        let nproc = num_cpus::get();
-        proc.start(superuser, pg_port(), nproc).await?;
-        wait_until_ready(restore_connstring.clone(), "neondb".to_string()).await;
-
-        Some(proc)
-    } else {
-        info!("restore_connection_string specified, not running postgres process");
-        None
+    let pgbin = pg_bin_dir.join("postgres");
+    let pg_version = match get_pg_version(pgbin.as_ref()) {
+        PostgresMajorVersion::V14 => 14,
+        PostgresMajorVersion::V15 => 15,
+        PostgresMajorVersion::V16 => 16,
+        PostgresMajorVersion::V17 => 17,
     };
+    let superuser = "cloud_admin"; // XXX: this shouldn't be hard-coded
+    postgres_initdb::do_run_initdb(postgres_initdb::RunInitdbArgs {
+        superuser,
+        locale: DEFAULT_LOCALE, // XXX: this shouldn't be hard-coded,
+        pg_version,
+        initdb_bin: pg_bin_dir.join("initdb").as_ref(),
+        library_search_path: &pg_lib_dir, // TODO: is this right? Prob works in compute image, not sure about neon_local.
+        pgdata: &pgdata_dir,
+    })
+    .await
+    .context("initdb")?;
+
+    let nproc = num_cpus::get();
+
+    //
+    // Launch postgres process
+    //
+    let mut postgres_proc = tokio::process::Command::new(pgbin)
+        .arg("-D")
+        .arg(&pgdata_dir)
+        .args(["-p", &format!("{pg_port}")])
+        .args(["-c", "wal_level=minimal"])
+        .args(["-c", "shared_buffers=10GB"])
+        .args(["-c", "max_wal_senders=0"])
+        .args(["-c", "fsync=off"])
+        .args(["-c", "full_page_writes=off"])
+        .args(["-c", "synchronous_commit=off"])
+        .args(["-c", "maintenance_work_mem=8388608"])
+        .args(["-c", &format!("max_parallel_maintenance_workers={nproc}")])
+        .args(["-c", &format!("max_parallel_workers={nproc}")])
+        .args(["-c", &format!("max_parallel_workers_per_gather={nproc}")])
+        .args(["-c", &format!("max_worker_processes={nproc}")])
+        .args([
+            "-c",
+            &format!(
+                "effective_io_concurrency={}",
+                if cfg!(target_os = "macos") { 0 } else { 100 }
+            ),
+        ])
+        .env_clear()
+        .env("LD_LIBRARY_PATH", &pg_lib_dir)
+        .stdout(std::process::Stdio::piped())
+        .stderr(std::process::Stdio::piped())
+        .spawn()
+        .context("spawn postgres")?;
+
+    info!("spawned postgres, waiting for it to become ready");
+    tokio::spawn(
+        child_stdio_to_log::relay_process_output(
+            postgres_proc.stdout.take(),
+            postgres_proc.stderr.take(),
+        )
+        .instrument(info_span!("postgres")),
+    );
+
+    // Create neondb database in the running postgres
+    let restore_pg_connstring =
+        format!("host=localhost port={pg_port} user={superuser} dbname=postgres");
+
+    let start_time = std::time::Instant::now();
+
+    loop {
+        if start_time.elapsed() > PG_WAIT_TIMEOUT {
+            error!(
+                "timeout exceeded: failed to poll postgres and create database within 10 minutes"
+            );
+            std::process::exit(1);
+        }
+
+        match tokio_postgres::connect(&restore_pg_connstring, tokio_postgres::NoTls).await {
+            Ok((client, connection)) => {
+                // Spawn the connection handling task to maintain the connection
+                tokio::spawn(async move {
+                    if let Err(e) = connection.await {
+                        warn!("connection error: {}", e);
+                    }
+                });
+
+                match client.simple_query("CREATE DATABASE neondb;").await {
+                    Ok(_) => {
+                        info!("created neondb database");
+                        break;
+                    }
+                    Err(e) => {
+                        warn!(
+                            "failed to create database: {}, retying in {}s",
+                            e,
+                            PG_WAIT_RETRY_INTERVAL.as_secs_f32()
+                        );
+                        tokio::time::sleep(PG_WAIT_RETRY_INTERVAL).await;
+                        continue;
+                    }
+                }
+            }
+            Err(_) => {
+                info!(
+                    "postgres not ready yet, retrying in {}s",
+                    PG_WAIT_RETRY_INTERVAL.as_secs_f32()
+                );
+                tokio::time::sleep(PG_WAIT_RETRY_INTERVAL).await;
+                continue;
+            }
+        }
+    }
+
+    let restore_pg_connstring = restore_pg_connstring.replace("dbname=postgres", "dbname=neondb");
+
+    let dumpdir = working_directory.join("dumpdir");
 
     let common_args = [
         // schema mapping (prob suffices to specify them on one side)
@@ -416,66 +309,64 @@ pub(crate) async fn main() -> anyhow::Result<()> {
         "--no-tablespaces".to_string(),
         // format
         "--format".to_string(),
-        "custom".to_string(),
+        "directory".to_string(),
         // concurrency
-        // "--jobs".to_string(),
-        // num_cpus::get().to_string(),
+        "--jobs".to_string(),
+        num_cpus::get().to_string(),
         // progress updates
         "--verbose".to_string(),
     ];
 
     info!("dump into the working directory");
-    let mut pg_dump = tokio::process::Command::new(pg_bin_dir.join("pg_dump"))
-        .args(&common_args)
-        .arg("--no-sync")
-        // POSITIONAL args
-        // source db (db name included in connection string)
-        .arg(&source_connstring)
-        // how we run it
-        .env_clear()
-        .env("LD_LIBRARY_PATH", &pg_lib_dir)
-        .kill_on_drop(true)
-        .stdout(std::process::Stdio::piped())
-        .stderr(std::process::Stdio::piped())
-        .spawn()
-        .context("spawn pg_dump")?;
-
-    const BUF_SIZE: usize = 64 * 1024 * 1024;
-    let mut buf = BufReader::with_capacity(BUF_SIZE, pg_dump.stdout.take().unwrap());
-
-    tokio::spawn(async move {
-        info!(pid=%pg_dump.id().unwrap(), "spawned pg_dump");
-        let st = pg_dump.wait().await.expect("wait for pg_dump");
-        info!(status=?st, "pg_dump exited");
-        if !st.success() {
-            warn!(status=%st, "pg_dump failed, restore will likely fail as well");
-        }
-
-        tokio::spawn(
-            child_stdio_to_log::relay_process_output(
-                None,
-                pg_dump.stderr.take(),
-            )
-            .instrument(info_span!("pg_dump")),
-        );
-    });
-
-    // TODO: do it in a streaming way, plenty of internal research done on this already
-    // TODO: do the unlogged table trick
     {
-        let mut pg_restore = tokio::process::Command::new(pg_bin_dir.join("pg_restore"))
+        let mut pg_dump = tokio::process::Command::new(pg_bin_dir.join("pg_dump"))
             .args(&common_args)
-            .arg("-d")
-            .arg(&restore_connstring)
-            .arg("--clean")
-            .arg("--if-exists")
+            .arg("-f")
+            .arg(&dumpdir)
+            .arg("--no-sync")
+            // POSITIONAL args
+            // source db (db name included in connection string)
+            .arg(&source_connection_string)
+            // how we run it
+            .env_clear()
+            .env("LD_LIBRARY_PATH", &pg_lib_dir)
+            .kill_on_drop(true)
+            .stdout(std::process::Stdio::piped())
+            .stderr(std::process::Stdio::piped())
+            .spawn()
+            .context("spawn pg_dump")?;
+
+        info!(pid=%pg_dump.id().unwrap(), "spawned pg_dump");
+
+        tokio::spawn(
+            child_stdio_to_log::relay_process_output(pg_dump.stdout.take(), pg_dump.stderr.take())
+                .instrument(info_span!("pg_dump")),
+        );
+
+        let st = pg_dump.wait().await.context("wait for pg_dump")?;
+        info!(status=?st, "pg_dump exited");
+        if !st.success() {
+            warn!(status=%st, "pg_dump failed, restore will likely fail as well");
+        }
+    }
+
+    // TODO: do it in a streaming way, plenty of internal research done on this already
+    // TODO: do the unlogged table trick
+
+    info!("restore from working directory into vanilla postgres");
+    {
+        let mut pg_restore = tokio::process::Command::new(pg_bin_dir.join("pg_restore"))
+            .args(&common_args)
+            .arg("-d")
+            .arg(&restore_pg_connstring)
+            // POSITIONAL args
+            .arg(&dumpdir)
             // how we run it
             .env_clear()
             .env("LD_LIBRARY_PATH", &pg_lib_dir)
             .kill_on_drop(true)
             .stdout(std::process::Stdio::piped())
             .stderr(std::process::Stdio::piped())
-            .stdin(std::process::Stdio::piped())
             .spawn()
             .context("spawn pg_restore")?;
 
@@ -487,48 +378,51 @@ pub(crate) async fn main() -> anyhow::Result<()> {
             )
             .instrument(info_span!("pg_restore")),
         );
-
-        let mut restore_stdin = pg_restore.stdin.take().unwrap();
-
-        tokio::spawn(async move {
-            tokio::io::copy_buf(&mut buf, &mut restore_stdin).await.expect("pg_restore failed to read from pg_dump");
-        });
-
         let st = pg_restore.wait().await.context("wait for pg_restore")?;
         info!(status=?st, "pg_restore exited");
         if !st.success() {
             warn!(status=%st, "pg_restore failed, restore will likely fail as well");
         }
-
     }
 
-    if let Some(mut proc) = postgres_proc {
-        // If interactive mode, wait for Ctrl+C
-        if args.interactive {
-            info!("Running in interactive mode. Press Ctrl+C to shut down.");
-            tokio::signal::ctrl_c().await.context("wait for ctrl-c")?;
-        }
+    // If interactive mode, wait for Ctrl+C
+    if args.interactive {
+        info!("Running in interactive mode. Press Ctrl+C to shut down.");
+        tokio::signal::ctrl_c().await.context("wait for ctrl-c")?;
+    }
 
-        proc.shutdown().await?;
+    info!("shutdown postgres");
+    {
+        nix::sys::signal::kill(
+            Pid::from_raw(
+                i32::try_from(postgres_proc.id().unwrap()).expect("convert child pid to i32"),
+            ),
+            nix::sys::signal::SIGTERM,
+        )
+        .context("signal postgres to shut down")?;
+        postgres_proc
+            .wait()
+            .await
+            .context("wait for postgres to shut down")?;
+    }
 
-        // Only sync if s3_prefix was specified
-        if let Some(s3_prefix) = args.s3_prefix {
-            info!("upload pgdata");
-            aws_s3_sync::sync(Utf8Path::new(&pgdata_dir), &s3_prefix.append("/pgdata/"))
+    // Only sync if s3_prefix was specified
+    if let Some(s3_prefix) = args.s3_prefix {
+        info!("upload pgdata");
+        aws_s3_sync::sync(Utf8Path::new(&pgdata_dir), &s3_prefix.append("/pgdata/"))
+            .await
+            .context("sync dump directory to destination")?;
+
+        info!("write status");
+        {
+            let status_dir = working_directory.join("status");
+            std::fs::create_dir(&status_dir).context("create status directory")?;
+            let status_file = status_dir.join("pgdata");
+            std::fs::write(&status_file, serde_json::json!({"done": true}).to_string())
+                .context("write status file")?;
+            aws_s3_sync::sync(&status_dir, &s3_prefix.append("/status/"))
                 .await
-                .context("sync dump directory to destination")?;
-
-            info!("write status");
-            {
-                let status_dir = working_directory.join("status");
-                std::fs::create_dir(&status_dir).context("create status directory")?;
-                let status_file = status_dir.join("pgdata");
-                std::fs::write(&status_file, serde_json::json!({"done": true}).to_string())
-                    .context("write status file")?;
-                aws_s3_sync::sync(&status_dir, &s3_prefix.append("/status/"))
-                    .await
-                    .context("sync status directory to destination")?;
-            }
+                .context("sync status directory to destination")?;
         }
     }
 

compute_tools/src/compute.rs

@@ -41,14 +41,14 @@ use crate::local_proxy;
 use crate::pg_helpers::*;
 use crate::spec::*;
 use crate::spec_apply::ApplySpecPhase::{
-    CreateAndAlterDatabases, CreateAndAlterRoles, CreateAvailabilityCheck, CreateSuperUser,
-    DropInvalidDatabases, DropRoles, HandleNeonExtension, HandleOtherExtensions,
-    RenameAndDeleteDatabases, RenameRoles, RunInEachDatabase,
+    CreateAndAlterDatabases, CreateAndAlterRoles, CreateAvailabilityCheck, CreateSchemaNeon,
+    CreateSuperUser, DropInvalidDatabases, DropRoles, FinalizeDropLogicalSubscriptions,
+    HandleNeonExtension, HandleOtherExtensions, RenameAndDeleteDatabases, RenameRoles,
+    RunInEachDatabase,
 };
 use crate::spec_apply::PerDatabasePhase;
 use crate::spec_apply::PerDatabasePhase::{
-    ChangeSchemaPerms, DeleteDBRoleReferences, DropSubscriptionsForDeletedDatabases,
-    HandleAnonExtension,
+    ChangeSchemaPerms, DeleteDBRoleReferences, DropLogicalSubscriptions, HandleAnonExtension,
 };
 use crate::spec_apply::{apply_operations, MutableApplyContext, DB};
 use crate::sync_sk::{check_if_synced, ping_safekeeper};
@@ -340,6 +340,15 @@ impl ComputeNode {
         self.state.lock().unwrap().status
     }
 
+    pub fn get_timeline_id(&self) -> Option<TimelineId> {
+        self.state
+            .lock()
+            .unwrap()
+            .pspec
+            .as_ref()
+            .map(|s| s.timeline_id)
+    }
+
     // Remove `pgdata` directory and create it again with right permissions.
     fn create_pgdata(&self) -> Result<()> {
         // Ignore removal error, likely it is a 'No such file or directory (os error 2)'.
@@ -929,6 +938,48 @@ impl ComputeNode {
                 .map(|role| (role.name.clone(), role))
                 .collect::<HashMap<String, Role>>();
 
+            // Check if we need to drop subscriptions before starting the endpoint.
+            //
+            // It is important to do this operation exactly once when endpoint starts on a new branch.
+            // Otherwise, we may drop not inherited, but newly created subscriptions.
+            //
+            // We cannot rely only on spec.drop_subscriptions_before_start flag,
+            // because if for some reason compute restarts inside VM,
+            // it will start again with the same spec and flag value.
+            //
+            // To handle this, we save the fact of the operation in the database
+            // in the neon.drop_subscriptions_done table.
+            // If the table does not exist, we assume that the operation was never performed, so we must do it.
+            // If table exists, we check if the operation was performed on the current timelilne.
+            //
+            let mut drop_subscriptions_done = false;
+
+            if spec.drop_subscriptions_before_start {
+                let timeline_id = self.get_timeline_id().context("timeline_id must be set")?;
+                let query = format!("select 1 from neon.drop_subscriptions_done where timeline_id = '{}'", timeline_id);
+
+                info!("Checking if drop subscription operation was already performed for timeline_id: {}", timeline_id);
+
+                drop_subscriptions_done =  match
+                    client.simple_query(&query).await {
+                    Ok(result) => {
+                        matches!(&result[0], postgres::SimpleQueryMessage::Row(_))
+                    },
+                    Err(e) =>
+                    {
+                        match e.code() {
+                            Some(&SqlState::UNDEFINED_TABLE) => false,
+                            _ => {
+                                // We don't expect any other error here, except for the schema/table not existing
+                                error!("Error checking if drop subscription operation was already performed: {}", e);
+                                return Err(e.into());
+                            }
+                        }
+                    }
+                }
+            };
+
+
             let jwks_roles = Arc::new(
                 spec.as_ref()
                     .local_proxy_config
@@ -996,7 +1047,7 @@ impl ComputeNode {
                         jwks_roles.clone(),
                         concurrency_token.clone(),
                         db,
-                        [DropSubscriptionsForDeletedDatabases].to_vec(),
+                        [DropLogicalSubscriptions].to_vec(),
                     );
 
                     Ok(spawn(fut))
@@ -1024,6 +1075,7 @@ impl ComputeNode {
                 CreateAndAlterRoles,
                 RenameAndDeleteDatabases,
                 CreateAndAlterDatabases,
+                CreateSchemaNeon,
             ] {
                 info!("Applying phase {:?}", &phase);
                 apply_operations(
@@ -1064,6 +1116,17 @@ impl ComputeNode {
                     }
 
                     let conf = Arc::new(conf);
+                    let mut phases = vec![
+                        DeleteDBRoleReferences,
+                        ChangeSchemaPerms,
+                        HandleAnonExtension,
+                    ];
+
+                    if spec.drop_subscriptions_before_start && !drop_subscriptions_done {
+                        info!("Adding DropLogicalSubscriptions phase because drop_subscriptions_before_start is set");
+                        phases.push(DropLogicalSubscriptions);
+                    }
+
                     let fut = Self::apply_spec_sql_db(
                         spec.clone(),
                         conf,
@@ -1071,12 +1134,7 @@ impl ComputeNode {
                         jwks_roles.clone(),
                         concurrency_token.clone(),
                         db,
-                        [
-                            DeleteDBRoleReferences,
-                            ChangeSchemaPerms,
-                            HandleAnonExtension,
-                        ]
-                        .to_vec(),
+                        phases,
                     );
 
                     Ok(spawn(fut))
@@ -1088,12 +1146,20 @@ impl ComputeNode {
                 handle.await??;
             }
 
-            for phase in vec![
+            let mut phases = vec![
                 HandleOtherExtensions,
-                HandleNeonExtension,
+                HandleNeonExtension, // This step depends on CreateSchemaNeon
                 CreateAvailabilityCheck,
                 DropRoles,
-            ] {
+            ];
+
+            // This step depends on CreateSchemaNeon
+            if spec.drop_subscriptions_before_start && !drop_subscriptions_done {
+                info!("Adding FinalizeDropLogicalSubscriptions phase because drop_subscriptions_before_start is set");
+                phases.push(FinalizeDropLogicalSubscriptions);
+            }
+
+            for phase in phases {
                 debug!("Applying phase {:?}", &phase);
                 apply_operations(
                     spec.clone(),
@@ -1420,6 +1486,7 @@ impl ComputeNode {
             // First, create control files for all availale extensions
             extension_server::create_control_files(remote_extensions, &self.pgbin);
 
+            // Second, preload all remote extensions specified in the shared_preload_libraries
             let library_load_start_time = Utc::now();
             let remote_ext_metrics = self.prepare_preload_libraries(&pspec.spec)?;
 
@@ -1463,6 +1530,14 @@ impl ComputeNode {
                         Ok(())
                     },
                 )?;
+
+                let postgresql_conf_path = pgdata_path.join("postgresql.conf");
+                if config::line_in_file(
+                    &postgresql_conf_path,
+                    "neon.disable_logical_replication_subscribers=false",
+                )? {
+                    info!("updated postgresql.conf to set neon.disable_logical_replication_subscribers=false");
+                }
                 self.pg_reload_conf()?;
             }
             self.post_apply_config()?;
@@ -1834,8 +1909,9 @@ LIMIT 100",
             .as_ref()
             .ok_or(anyhow::anyhow!("Remote extensions are not configured"))?;
 
-        info!("parse shared_preload_libraries from spec.cluster.settings");
         let mut libs_vec = Vec::new();
+
+        info!("parse shared_preload_libraries from spec.cluster.settings");
         if let Some(libs) = spec.cluster.settings.find("shared_preload_libraries") {
             libs_vec = libs
                 .split(&[',', '\'', ' '])
@@ -1843,9 +1919,9 @@ LIMIT 100",
                 .map(str::to_string)
                 .collect();
         }
-        info!("parse shared_preload_libraries from provided postgresql.conf");
 
-        // that is used in neon_local and python tests
+        // This is used in neon_local and python tests
+        info!("parse shared_preload_libraries from provided postgresql.conf");
         if let Some(conf) = &spec.cluster.postgresql_conf {
             let conf_lines = conf.split('\n').collect::<Vec<&str>>();
             let mut shared_preload_libraries_line = "";
@@ -1869,7 +1945,10 @@ LIMIT 100",
         // Assume that they are already present locally.
         libs_vec.retain(|lib| remote_extensions.library_index.contains_key(lib));
 
-        info!("Downloading to shared preload libraries: {:?}", &libs_vec);
+        info!(
+            "Downloading extensions specified in shared_preload_libraries: {:?}",
+            &libs_vec
+        );
 
         let mut download_tasks = Vec::new();
         for library in &libs_vec {

compute_tools/src/config.rs

@@ -129,6 +129,13 @@ pub fn write_postgres_conf(
 
     writeln!(file, "neon.extension_server_port={}", extension_server_port)?;
 
+    if spec.drop_subscriptions_before_start {
+        writeln!(file, "neon.disable_logical_replication_subscribers=true")?;
+    } else {
+        // be explicit about the default value
+        writeln!(file, "neon.disable_logical_replication_subscribers=false")?;
+    }
+
     // This is essential to keep this line at the end of the file,
     // because it is intended to override any settings above.
     writeln!(file, "include_if_exists = 'compute_ctl_temp_override.conf'")?;

compute_tools/src/extension_server.rs

@@ -148,18 +148,18 @@ fn parse_pg_version(human_version: &str) -> PostgresMajorVersion {
         },
         _ => {}
     }
-    panic!("Unsuported postgres version {human_version}");
+    panic!("Unsupported Postgres version {human_version}");
 }
 
-// download the archive for a given extension,
-// unzip it, and place files in the appropriate locations (share/lib)
+/// Download the archive for a given extension,
+/// unzip it, and place files in the appropriate locations (share/lib)
 pub async fn download_extension(
     ext_name: &str,
     ext_path: &RemotePath,
     ext_remote_storage: &str,
     pgbin: &str,
 ) -> Result<u64> {
-    info!("Download extension {:?} from {:?}", ext_name, ext_path);
+    info!("Downloading extension {:?} from {:?}", ext_name, ext_path);
 
     // TODO add retry logic
     let download_buffer =
@@ -200,23 +200,23 @@ pub async fn download_extension(
     // move contents of the libdir / sharedir in unzipped archive to the correct local paths
     for paths in [sharedir_paths, libdir_paths] {
         let (zip_dir, real_dir) = paths;
-        info!("mv {zip_dir:?}/*  {real_dir:?}");
+        info!("Moving {zip_dir:?}/* to {real_dir:?}");
         for file in std::fs::read_dir(zip_dir)? {
             let old_file = file?.path();
             let new_file =
                 Path::new(&real_dir).join(old_file.file_name().context("error parsing file")?);
-            info!("moving {old_file:?} to {new_file:?}");
+            info!("Moving {old_file:?} to {new_file:?}");
 
             // extension download failed: Directory not empty (os error 39)
             match std::fs::rename(old_file, new_file) {
-                Ok(()) => info!("move succeeded"),
+                Ok(()) => info!("Move succeeded"),
                 Err(e) => {
-                    warn!("move failed, probably because the extension already exists: {e}")
+                    warn!("Move failed, probably because the extension already exists: {e}")
                 }
             }
         }
     }
-    info!("done moving extension {ext_name}");
+    info!("Done moving extension {ext_name}");
     Ok(download_size)
 }
 
@@ -239,10 +239,16 @@ pub fn create_control_files(remote_extensions: &RemoteExtSpec, pgbin: &str) {
         for (control_name, control_content) in &ext_data.control_data {
             let control_path = local_sharedir.join(control_name);
             if !control_path.exists() {
-                info!("writing file {:?}{:?}", control_path, control_content);
+                info!(
+                    "Writing control file content {:?}: {:?}",
+                    control_path, control_content
+                );
                 std::fs::write(control_path, control_content).unwrap();
             } else {
-                warn!("control file {:?} exists both locally and remotely. ignoring the remote version.", control_path);
+                warn!(
+                    "Control file {:?} exists locally. Ignoring the version from the spec.",
+                    control_path
+                );
             }
         }
     }
@@ -250,9 +256,7 @@ pub fn create_control_files(remote_extensions: &RemoteExtSpec, pgbin: &str) {
 
 // Do request to extension storage proxy, i.e.
 // curl http://pg-ext-s3-gateway/latest/v15/extensions/anon.tar.zst
-// using HHTP GET
-// and return the response body as bytes
-//
+// using HTTP GET and return the response body as bytes.
 async fn download_extension_tar(ext_remote_storage: &str, ext_path: &str) -> Result<Bytes> {
     let uri = format!("{}/{}", ext_remote_storage, ext_path);
 

compute_tools/src/spec_apply.rs

@@ -47,7 +47,7 @@ pub enum PerDatabasePhase {
     DeleteDBRoleReferences,
     ChangeSchemaPerms,
     HandleAnonExtension,
-    DropSubscriptionsForDeletedDatabases,
+    DropLogicalSubscriptions,
 }
 
 #[derive(Clone, Debug)]
@@ -58,11 +58,13 @@ pub enum ApplySpecPhase {
     CreateAndAlterRoles,
     RenameAndDeleteDatabases,
     CreateAndAlterDatabases,
+    CreateSchemaNeon,
     RunInEachDatabase { db: DB, subphase: PerDatabasePhase },
     HandleOtherExtensions,
     HandleNeonExtension,
     CreateAvailabilityCheck,
     DropRoles,
+    FinalizeDropLogicalSubscriptions,
 }
 
 pub struct Operation {
@@ -331,7 +333,7 @@ async fn get_operations<'a>(
                             // NB: there could be other db states, which prevent us from dropping
                             // the database. For example, if db is used by any active subscription
                             // or replication slot.
-                            // Such cases are handled in the DropSubscriptionsForDeletedDatabases
+                            // Such cases are handled in the DropLogicalSubscriptions
                             // phase. We do all the cleanup before actually dropping the database.
                             let drop_db_query: String = format!(
                                 "DROP DATABASE IF EXISTS {} WITH (FORCE)",
@@ -442,13 +444,19 @@ async fn get_operations<'a>(
 
             Ok(Box::new(operations))
         }
+        ApplySpecPhase::CreateSchemaNeon => Ok(Box::new(once(Operation {
+            query: String::from("CREATE SCHEMA IF NOT EXISTS neon"),
+            comment: Some(String::from(
+                "create schema for neon extension and utils tables",
+            )),
+        }))),
         ApplySpecPhase::RunInEachDatabase { db, subphase } => {
             match subphase {
-                PerDatabasePhase::DropSubscriptionsForDeletedDatabases => {
+                PerDatabasePhase::DropLogicalSubscriptions => {
                     match &db {
                         DB::UserDB(db) => {
                             let drop_subscription_query: String = format!(
-                                include_str!("sql/drop_subscription_for_drop_dbs.sql"),
+                                include_str!("sql/drop_subscriptions.sql"),
                                 datname_str = escape_literal(&db.name),
                             );
 
@@ -666,10 +674,6 @@ async fn get_operations<'a>(
         }
         ApplySpecPhase::HandleNeonExtension => {
             let operations = vec![
-                Operation {
-                    query: String::from("CREATE SCHEMA IF NOT EXISTS neon"),
-                    comment: Some(String::from("init: add schema for extension")),
-                },
                 Operation {
                     query: String::from("CREATE EXTENSION IF NOT EXISTS neon WITH SCHEMA neon"),
                     comment: Some(String::from(
@@ -712,5 +716,9 @@ async fn get_operations<'a>(
 
             Ok(Box::new(operations))
         }
+        ApplySpecPhase::FinalizeDropLogicalSubscriptions => Ok(Box::new(once(Operation {
+            query: String::from(include_str!("sql/finalize_drop_subscriptions.sql")),
+            comment: None,
+        }))),
     }
 }

compute_tools/src/sql/finalize_drop_subscriptions.sql

@@ -0,0 +1,21 @@
+DO $$
+BEGIN
+    IF NOT EXISTS(
+        SELECT 1
+        FROM pg_catalog.pg_tables
+        WHERE tablename = 'drop_subscriptions_done'
+        AND schemaname = 'neon'
+    )
+    THEN
+        CREATE TABLE neon.drop_subscriptions_done
+        (id serial primary key, timeline_id text);
+    END IF;
+
+    -- preserve the timeline_id of the last drop_subscriptions run
+    -- to ensure that the cleanup of a timeline is executed only once.
+    -- use upsert to avoid the table bloat in case of cascade branching (branch of a branch)
+    INSERT INTO neon.drop_subscriptions_done VALUES (1, current_setting('neon.timeline_id'))
+    ON CONFLICT (id) DO UPDATE
+    SET timeline_id = current_setting('neon.timeline_id');
+END
+$$

control_plane/README.md

@@ -1,6 +1,10 @@
-# Control Plane and Neon Local
+# Local Development Control Plane (`neon_local`)
 
-This crate contains tools to start a Neon development environment locally. This utility can be used with the `cargo neon` command.
+This crate contains tools to start a Neon development environment locally. This utility can be used with the `cargo neon` command.  This is a convenience to invoke
+the `neon_local` binary.
+
+**Note**: this is a dev/test tool -- a minimal control plane suitable for testing
+code changes locally, but not suitable for running production systems.
 
 ## Example: Start with Postgres 16
 

control_plane/src/bin/neon_local.rs

@@ -1357,6 +1357,7 @@ async fn handle_endpoint(subcmd: &EndpointCmd, env: &local_env::LocalEnv) -> Res
                 args.pg_version,
                 mode,
                 !args.update_catalog,
+                false,
             )?;
         }
         EndpointCmd::Start(args) => {

control_plane/src/endpoint.rs

@@ -76,6 +76,7 @@ pub struct EndpointConf {
     http_port: u16,
     pg_version: u32,
     skip_pg_catalog_updates: bool,
+    drop_subscriptions_before_start: bool,
     features: Vec<ComputeFeature>,
 }
 
@@ -143,6 +144,7 @@ impl ComputeControlPlane {
         pg_version: u32,
         mode: ComputeMode,
         skip_pg_catalog_updates: bool,
+        drop_subscriptions_before_start: bool,
     ) -> Result<Arc<Endpoint>> {
         let pg_port = pg_port.unwrap_or_else(|| self.get_port());
         let http_port = http_port.unwrap_or_else(|| self.get_port() + 1);
@@ -162,6 +164,7 @@ impl ComputeControlPlane {
             // with this we basically test a case of waking up an idle compute, where
             // we also skip catalog updates in the cloud.
             skip_pg_catalog_updates,
+            drop_subscriptions_before_start,
             features: vec![],
         });
 
@@ -177,6 +180,7 @@ impl ComputeControlPlane {
                 pg_port,
                 pg_version,
                 skip_pg_catalog_updates,
+                drop_subscriptions_before_start,
                 features: vec![],
             })?,
         )?;
@@ -240,6 +244,7 @@ pub struct Endpoint {
     // Optimizations
     skip_pg_catalog_updates: bool,
 
+    drop_subscriptions_before_start: bool,
     // Feature flags
     features: Vec<ComputeFeature>,
 }
@@ -291,6 +296,7 @@ impl Endpoint {
             tenant_id: conf.tenant_id,
             pg_version: conf.pg_version,
             skip_pg_catalog_updates: conf.skip_pg_catalog_updates,
+            drop_subscriptions_before_start: conf.drop_subscriptions_before_start,
             features: conf.features,
         })
     }
@@ -625,6 +631,7 @@ impl Endpoint {
             shard_stripe_size: Some(shard_stripe_size),
             local_proxy_config: None,
             reconfigure_concurrency: 1,
+            drop_subscriptions_before_start: self.drop_subscriptions_before_start,
         };
         let spec_path = self.endpoint_path().join("spec.json");
         std::fs::write(spec_path, serde_json::to_string_pretty(&spec)?)?;

control_plane/src/pageserver.rs

@@ -352,6 +352,16 @@ impl PageServerNode {
                 .map(serde_json::from_str)
                 .transpose()
                 .context("Failed to parse 'compaction_algorithm' json")?,
+            l0_flush_delay_threshold: settings
+                .remove("l0_flush_delay_threshold")
+                .map(|x| x.parse::<usize>())
+                .transpose()
+                .context("Failed to parse 'l0_flush_delay_threshold' as an integer")?,
+            l0_flush_stall_threshold: settings
+                .remove("l0_flush_stall_threshold")
+                .map(|x| x.parse::<usize>())
+                .transpose()
+                .context("Failed to parse 'l0_flush_stall_threshold' as an integer")?,
             gc_horizon: settings
                 .remove("gc_horizon")
                 .map(|x| x.parse::<u64>())

docker-compose/docker-compose.yml

@@ -150,8 +150,8 @@ services:
         - REPOSITORY=${REPOSITORY:-neondatabase}
         - COMPUTE_IMAGE=compute-node-v${PG_VERSION:-16}
         - TAG=${TAG:-latest}
-        - http_proxy=$http_proxy
-        - https_proxy=$https_proxy
+        - http_proxy=${http_proxy:-}
+        - https_proxy=${https_proxy:-}
     environment:
       - PG_VERSION=${PG_VERSION:-16}
       #- RUST_BACKTRACE=1
@@ -185,6 +185,8 @@ services:
   neon-test-extensions:
     profiles: ["test-extensions"]
     image: ${REPOSITORY:-neondatabase}/neon-test-extensions-v${PG_TEST_VERSION:-16}:${TAG:-latest}
+    environment:
+      - PGPASSWORD=cloud_admin
     entrypoint:
       - "/bin/bash"
       - "-c"

docker-compose/docker_compose_test.sh

@@ -22,7 +22,6 @@ PSQL_OPTION="-h localhost -U cloud_admin -p 55433 -d postgres"
 cleanup() {
     echo "show container information"
     docker ps
-    docker compose --profile test-extensions -f $COMPOSE_FILE logs
     echo "stop containers..."
     docker compose --profile test-extensions -f $COMPOSE_FILE down
 }
@@ -41,7 +40,6 @@ for pg_version in ${TEST_VERSION_ONLY-14 15 16 17}; do
         cnt=`expr $cnt + 3`
         if [ $cnt -gt 60 ]; then
             echo "timeout before the compute is ready."
-            cleanup
             exit 1
         fi
         if docker compose --profile test-extensions -f $COMPOSE_FILE logs "compute_is_ready" | grep -q "accepting connections"; then
@@ -63,11 +61,9 @@ for pg_version in ${TEST_VERSION_ONLY-14 15 16 17}; do
         docker cp $TMPDIR/data $COMPUTE_CONTAINER_NAME:/ext-src/pg_hint_plan-src/
         rm -rf $TMPDIR
         # We are running tests now
-        if docker exec -e SKIP=timescaledb-src,rdkit-src,postgis-src,pgx_ulid-src,pgtap-src,pg_tiktoken-src,pg_jsonschema-src,pg_graphql-src,kq_imcx-src,wal2json_2_5-src \
+        if ! docker exec -e SKIP=timescaledb-src,rdkit-src,postgis-src,pgx_ulid-src,pgtap-src,pg_tiktoken-src,pg_jsonschema-src,pg_graphql-src,kq_imcx-src,wal2json_2_5-src \
             $TEST_CONTAINER_NAME /run-tests.sh | tee testout.txt
         then
-            cleanup
-        else
             FAILED=$(tail -1 testout.txt)
             for d in $FAILED
             do
@@ -77,9 +73,7 @@ for pg_version in ${TEST_VERSION_ONLY-14 15 16 17}; do
                 cat $d/regression.out $d/regression.diffs || true
             done
         rm -rf $FAILED
-        cleanup
         exit 1
         fi
     fi
-    cleanup
 done

docker-compose/run-tests.sh

@@ -7,7 +7,10 @@ LIST=$( (echo -e "${SKIP//","/"\n"}"; ls -d -- *-src) | sort | uniq -u)
 for d in ${LIST}
 do
        [ -d "${d}" ] || continue
-    psql -c "select 1" >/dev/null || break
+       if ! psql -w -c "select 1" >/dev/null; then
+          FAILED="${d} ${FAILED}"
+          break
+       fi
        USE_PGXS=1 make -C "${d}" installcheck || FAILED="${d} ${FAILED}"
 done
 [ -z "${FAILED}" ] && exit 0

libs/compute_api/src/spec.rs

@@ -138,6 +138,13 @@ pub struct ComputeSpec {
     /// enough spare connections for reconfiguration process to succeed.
     #[serde(default = "default_reconfigure_concurrency")]
     pub reconfigure_concurrency: usize,
+
+    /// If set to true, the compute_ctl will drop all subscriptions before starting the
+    /// compute. This is needed when we start an endpoint on a branch, so that child
+    /// would not compete with parent branch subscriptions
+    /// over the same replication content from publisher.
+    #[serde(default)] // Default false
+    pub drop_subscriptions_before_start: bool,
 }
 
 /// Feature flag to signal `compute_ctl` to enable certain experimental functionality.

libs/pageserver_api/src/config.rs

@@ -254,9 +254,18 @@ pub struct TenantConfigToml {
     // Duration::ZERO means automatic compaction is disabled.
     #[serde(with = "humantime_serde")]
     pub compaction_period: Duration,
-    // Level0 delta layer threshold for compaction.
+    /// Level0 delta layer threshold for compaction.
     pub compaction_threshold: usize,
     pub compaction_algorithm: crate::models::CompactionAlgorithmSettings,
+    /// Level0 delta layer threshold at which to delay layer flushes for compaction backpressure,
+    /// such that they take 2x as long, and start waiting for layer flushes during ephemeral layer
+    /// rolls. This helps compaction keep up with WAL ingestion, and avoids read amplification
+    /// blowing up. Should be >compaction_threshold. If None, defaults to 2 * compaction_threshold.
+    /// 0 to disable.
+    pub l0_flush_delay_threshold: Option<usize>,
+    /// Level0 delta layer threshold at which to stall layer flushes. 0 to disable. If None,
+    /// defaults to 4 * compaction_threshold. Must be >compaction_threshold to avoid deadlock.
+    pub l0_flush_stall_threshold: Option<usize>,
     // Determines how much history is retained, to allow
     // branching and read replicas at an older point in time.
     // The unit is #of bytes of WAL.
@@ -552,6 +561,8 @@ impl Default for TenantConfigToml {
             compaction_algorithm: crate::models::CompactionAlgorithmSettings {
                 kind: DEFAULT_COMPACTION_ALGORITHM,
             },
+            l0_flush_delay_threshold: None,
+            l0_flush_stall_threshold: None,
             gc_horizon: DEFAULT_GC_HORIZON,
             gc_period: humantime::parse_duration(DEFAULT_GC_PERIOD)
                 .expect("cannot parse default gc period"),

libs/pageserver_api/src/models.rs

@@ -462,6 +462,10 @@ pub struct TenantConfigPatch {
     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     pub compaction_algorithm: FieldPatch<CompactionAlgorithmSettings>,
     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
+    pub l0_flush_delay_threshold: FieldPatch<usize>,
+    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
+    pub l0_flush_stall_threshold: FieldPatch<usize>,
+    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     pub gc_horizon: FieldPatch<u64>,
     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     pub gc_period: FieldPatch<String>,
@@ -518,6 +522,8 @@ pub struct TenantConfig {
     pub compaction_threshold: Option<usize>,
     // defer parsing compaction_algorithm, like eviction_policy
     pub compaction_algorithm: Option<CompactionAlgorithmSettings>,
+    pub l0_flush_delay_threshold: Option<usize>,
+    pub l0_flush_stall_threshold: Option<usize>,
     pub gc_horizon: Option<u64>,
     pub gc_period: Option<String>,
     pub image_creation_threshold: Option<usize>,
@@ -551,6 +557,8 @@ impl TenantConfig {
             mut compaction_period,
             mut compaction_threshold,
             mut compaction_algorithm,
+            mut l0_flush_delay_threshold,
+            mut l0_flush_stall_threshold,
             mut gc_horizon,
             mut gc_period,
             mut image_creation_threshold,
@@ -583,6 +591,12 @@ impl TenantConfig {
         patch.compaction_period.apply(&mut compaction_period);
         patch.compaction_threshold.apply(&mut compaction_threshold);
         patch.compaction_algorithm.apply(&mut compaction_algorithm);
+        patch
+            .l0_flush_delay_threshold
+            .apply(&mut l0_flush_delay_threshold);
+        patch
+            .l0_flush_stall_threshold
+            .apply(&mut l0_flush_stall_threshold);
         patch.gc_horizon.apply(&mut gc_horizon);
         patch.gc_period.apply(&mut gc_period);
         patch
@@ -635,6 +649,8 @@ impl TenantConfig {
             compaction_period,
             compaction_threshold,
             compaction_algorithm,
+            l0_flush_delay_threshold,
+            l0_flush_stall_threshold,
             gc_horizon,
             gc_period,
             image_creation_threshold,

libs/remote_storage/src/azure_blob.rs

@@ -377,7 +377,8 @@ impl RemoteStorage for AzureBlobStorage {
 
                 let next_item = next_item?;
 
-                if timeout_try_cnt >= 2 {
+                // Log a warning if we saw two timeouts in a row before a successful request
+                if timeout_try_cnt > 2 {
                     tracing::warn!("Azure Blob Storage list timed out and succeeded after {} tries", timeout_try_cnt);
                 }
                 timeout_try_cnt = 1;

pageserver/src/metrics.rs

@@ -3,7 +3,7 @@ use metrics::{
     register_counter_vec, register_gauge_vec, register_histogram, register_histogram_vec,
     register_int_counter, register_int_counter_pair_vec, register_int_counter_vec,
     register_int_gauge, register_int_gauge_vec, register_uint_gauge, register_uint_gauge_vec,
-    Counter, CounterVec, Gauge, GaugeVec, Histogram, HistogramVec, IntCounter, IntCounterPair,
+    Counter, CounterVec, GaugeVec, Histogram, HistogramVec, IntCounter, IntCounterPair,
     IntCounterPairVec, IntCounterVec, IntGauge, IntGaugeVec, UIntGauge, UIntGaugeVec,
 };
 use once_cell::sync::Lazy;
@@ -38,6 +38,9 @@ pub(crate) enum StorageTimeOperation {
     #[strum(serialize = "layer flush")]
     LayerFlush,
 
+    #[strum(serialize = "layer flush delay")]
+    LayerFlushDelay,
+
     #[strum(serialize = "compact")]
     Compact,
 
@@ -395,15 +398,6 @@ pub(crate) static WAIT_LSN_TIME: Lazy<Histogram> = Lazy::new(|| {
     .expect("failed to define a metric")
 });
 
-static FLUSH_WAIT_UPLOAD_TIME: Lazy<GaugeVec> = Lazy::new(|| {
-    register_gauge_vec!(
-        "pageserver_flush_wait_upload_seconds",
-        "Time spent waiting for preceding uploads during layer flush",
-        &["tenant_id", "shard_id", "timeline_id"]
-    )
-    .expect("failed to define a metric")
-});
-
 static LAST_RECORD_LSN: Lazy<IntGaugeVec> = Lazy::new(|| {
     register_int_gauge_vec!(
         "pageserver_last_record_lsn",
@@ -2517,7 +2511,6 @@ impl Drop for AlwaysRecordingStorageTimeMetricsTimer {
 
 impl AlwaysRecordingStorageTimeMetricsTimer {
     /// Returns the elapsed duration of the timer.
-    #[allow(unused)]
     pub fn elapsed(&self) -> Duration {
         self.0.as_ref().expect("not dropped yet").elapsed()
     }
@@ -2575,7 +2568,7 @@ pub(crate) struct TimelineMetrics {
     shard_id: String,
     timeline_id: String,
     pub flush_time_histo: StorageTimeMetrics,
-    pub flush_wait_upload_time_gauge: Gauge,
+    pub flush_delay_histo: StorageTimeMetrics,
     pub compact_time_histo: StorageTimeMetrics,
     pub create_images_time_histo: StorageTimeMetrics,
     pub logical_size_histo: StorageTimeMetrics,
@@ -2621,9 +2614,12 @@ impl TimelineMetrics {
             &shard_id,
             &timeline_id,
         );
-        let flush_wait_upload_time_gauge = FLUSH_WAIT_UPLOAD_TIME
-            .get_metric_with_label_values(&[&tenant_id, &shard_id, &timeline_id])
-            .unwrap();
+        let flush_delay_histo = StorageTimeMetrics::new(
+            StorageTimeOperation::LayerFlushDelay,
+            &tenant_id,
+            &shard_id,
+            &timeline_id,
+        );
         let compact_time_histo = StorageTimeMetrics::new(
             StorageTimeOperation::Compact,
             &tenant_id,
@@ -2769,7 +2765,7 @@ impl TimelineMetrics {
             shard_id,
             timeline_id,
             flush_time_histo,
-            flush_wait_upload_time_gauge,
+            flush_delay_histo,
             compact_time_histo,
             create_images_time_histo,
             logical_size_histo,
@@ -2819,14 +2815,6 @@ impl TimelineMetrics {
         self.resident_physical_size_gauge.get()
     }
 
-    pub(crate) fn flush_wait_upload_time_gauge_add(&self, duration: f64) {
-        self.flush_wait_upload_time_gauge.add(duration);
-        crate::metrics::FLUSH_WAIT_UPLOAD_TIME
-            .get_metric_with_label_values(&[&self.tenant_id, &self.shard_id, &self.timeline_id])
-            .unwrap()
-            .add(duration);
-    }
-
     pub(crate) fn shutdown(&self) {
         let was_shutdown = self
             .shutdown
@@ -2844,7 +2832,6 @@ impl TimelineMetrics {
         let shard_id = &self.shard_id;
         let _ = LAST_RECORD_LSN.remove_label_values(&[tenant_id, shard_id, timeline_id]);
         let _ = DISK_CONSISTENT_LSN.remove_label_values(&[tenant_id, shard_id, timeline_id]);
-        let _ = FLUSH_WAIT_UPLOAD_TIME.remove_label_values(&[tenant_id, shard_id, timeline_id]);
         let _ = STANDBY_HORIZON.remove_label_values(&[tenant_id, shard_id, timeline_id]);
         {
             RESIDENT_PHYSICAL_SIZE_GLOBAL.sub(self.resident_physical_size_get());

pageserver/src/tenant.rs

@@ -5453,6 +5453,8 @@ pub(crate) mod harness {
                 compaction_period: Some(tenant_conf.compaction_period),
                 compaction_threshold: Some(tenant_conf.compaction_threshold),
                 compaction_algorithm: Some(tenant_conf.compaction_algorithm),
+                l0_flush_delay_threshold: tenant_conf.l0_flush_delay_threshold,
+                l0_flush_stall_threshold: tenant_conf.l0_flush_stall_threshold,
                 gc_horizon: Some(tenant_conf.gc_horizon),
                 gc_period: Some(tenant_conf.gc_period),
                 image_creation_threshold: Some(tenant_conf.image_creation_threshold),

pageserver/src/tenant/config.rs

@@ -281,6 +281,14 @@ pub struct TenantConfOpt {
     #[serde(default)]
     pub compaction_algorithm: Option<CompactionAlgorithmSettings>,
 
+    #[serde(skip_serializing_if = "Option::is_none")]
+    #[serde(default)]
+    pub l0_flush_delay_threshold: Option<usize>,
+
+    #[serde(skip_serializing_if = "Option::is_none")]
+    #[serde(default)]
+    pub l0_flush_stall_threshold: Option<usize>,
+
     #[serde(skip_serializing_if = "Option::is_none")]
     #[serde(default)]
     pub gc_horizon: Option<u64>,
@@ -394,6 +402,12 @@ impl TenantConfOpt {
                 .as_ref()
                 .unwrap_or(&global_conf.compaction_algorithm)
                 .clone(),
+            l0_flush_delay_threshold: self
+                .l0_flush_delay_threshold
+                .or(global_conf.l0_flush_delay_threshold),
+            l0_flush_stall_threshold: self
+                .l0_flush_stall_threshold
+                .or(global_conf.l0_flush_stall_threshold),
             gc_horizon: self.gc_horizon.unwrap_or(global_conf.gc_horizon),
             gc_period: self.gc_period.unwrap_or(global_conf.gc_period),
             image_creation_threshold: self
@@ -458,6 +472,8 @@ impl TenantConfOpt {
             mut compaction_period,
             mut compaction_threshold,
             mut compaction_algorithm,
+            mut l0_flush_delay_threshold,
+            mut l0_flush_stall_threshold,
             mut gc_horizon,
             mut gc_period,
             mut image_creation_threshold,
@@ -496,6 +512,12 @@ impl TenantConfOpt {
             .apply(&mut compaction_period);
         patch.compaction_threshold.apply(&mut compaction_threshold);
         patch.compaction_algorithm.apply(&mut compaction_algorithm);
+        patch
+            .l0_flush_delay_threshold
+            .apply(&mut l0_flush_delay_threshold);
+        patch
+            .l0_flush_stall_threshold
+            .apply(&mut l0_flush_stall_threshold);
         patch.gc_horizon.apply(&mut gc_horizon);
         patch
             .gc_period
@@ -566,6 +588,8 @@ impl TenantConfOpt {
             compaction_period,
             compaction_threshold,
             compaction_algorithm,
+            l0_flush_delay_threshold,
+            l0_flush_stall_threshold,
             gc_horizon,
             gc_period,
             image_creation_threshold,
@@ -623,6 +647,8 @@ impl From<TenantConfOpt> for models::TenantConfig {
             compaction_target_size: value.compaction_target_size,
             compaction_period: value.compaction_period.map(humantime),
             compaction_threshold: value.compaction_threshold,
+            l0_flush_delay_threshold: value.l0_flush_delay_threshold,
+            l0_flush_stall_threshold: value.l0_flush_stall_threshold,
             gc_horizon: value.gc_horizon,
             gc_period: value.gc_period.map(humantime),
             image_creation_threshold: value.image_creation_threshold,

pageserver/src/tenant/timeline.rs

@@ -22,11 +22,11 @@ use enumset::EnumSet;
 use fail::fail_point;
 use futures::{stream::FuturesUnordered, StreamExt};
 use handle::ShardTimelineId;
+use layer_manager::Shutdown;
 use offload::OffloadError;
 use once_cell::sync::Lazy;
 use pageserver_api::models::PageTraceEvent;
 use pageserver_api::{
-    config::tenant_conf_defaults::DEFAULT_COMPACTION_THRESHOLD,
     key::{
         KEY_SIZE, METADATA_KEY_BEGIN_PREFIX, METADATA_KEY_END_PREFIX, NON_INHERITED_RANGE,
         SPARSE_RANGE,
@@ -60,20 +60,14 @@ use utils::{
 };
 use wal_decoder::serialized_batch::{SerializedValueBatch, ValueMeta};
 
-use std::sync::atomic::Ordering as AtomicOrdering;
-use std::sync::OnceLock;
-use std::sync::{Arc, Mutex, RwLock, Weak};
+use std::array;
+use std::cmp::{max, min};
+use std::collections::btree_map::Entry;
+use std::collections::{BTreeMap, HashMap, HashSet};
+use std::ops::{ControlFlow, Deref, Range};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering as AtomicOrdering};
+use std::sync::{Arc, Mutex, OnceLock, RwLock, Weak};
 use std::time::{Duration, Instant, SystemTime};
-use std::{
-    array,
-    collections::{BTreeMap, HashMap, HashSet},
-    sync::atomic::AtomicU64,
-};
-use std::{cmp::min, ops::ControlFlow};
-use std::{
-    collections::btree_map::Entry,
-    ops::{Deref, Range},
-};
 
 use crate::l0_flush::{self, L0FlushGlobalState};
 use crate::{
@@ -150,19 +144,15 @@ use self::layer_manager::LayerManager;
 use self::logical_size::LogicalSize;
 use self::walreceiver::{WalReceiver, WalReceiverConf};
 
+use super::config::TenantConf;
+use super::remote_timeline_client::index::IndexPart;
+use super::remote_timeline_client::RemoteTimelineClient;
+use super::secondary::heatmap::{HeatMapLayer, HeatMapTimeline};
+use super::storage_layer::{LayerFringe, LayerVisibilityHint, ReadableLayer};
+use super::upload_queue::NotInitialized;
+use super::GcError;
 use super::{
-    config::TenantConf, storage_layer::LayerVisibilityHint, upload_queue::NotInitialized,
-    MaybeOffloaded,
-};
-use super::{debug_assert_current_span_has_tenant_and_timeline_id, AttachedTenantConf};
-use super::{remote_timeline_client::index::IndexPart, storage_layer::LayerFringe};
-use super::{
-    remote_timeline_client::RemoteTimelineClient, remote_timeline_client::WaitCompletionError,
-    storage_layer::ReadableLayer,
-};
-use super::{
-    secondary::heatmap::{HeatMapLayer, HeatMapTimeline},
-    GcError,
+    debug_assert_current_span_has_tenant_and_timeline_id, AttachedTenantConf, MaybeOffloaded,
 };
 
 #[cfg(test)]
@@ -408,6 +398,9 @@ pub struct Timeline {
     /// Timeline deletion will acquire both compaction and gc locks in whatever order.
     compaction_lock: tokio::sync::Mutex<()>,
 
+    /// If true, the last compaction failed.
+    compaction_failed: AtomicBool,
+
     /// Make sure we only have one running gc at a time.
     ///
     /// Must only be taken in two places:
@@ -1702,13 +1695,27 @@ impl Timeline {
             return Ok(false);
         }
 
-        match self.get_compaction_algorithm_settings().kind {
+        let result = match self.get_compaction_algorithm_settings().kind {
             CompactionAlgorithm::Tiered => {
                 self.compact_tiered(cancel, ctx).await?;
                 Ok(false)
             }
             CompactionAlgorithm::Legacy => self.compact_legacy(cancel, options, ctx).await,
-        }
+        };
+
+        // Signal compaction failure to avoid L0 flush stalls when it's broken.
+        match result {
+            Ok(_) => self.compaction_failed.store(false, AtomicOrdering::Relaxed),
+            Err(CompactionError::Other(_)) => {
+                self.compaction_failed.store(true, AtomicOrdering::Relaxed)
+            }
+            // Don't change the current value on offload failure or shutdown. We don't want to
+            // abruptly stall nor resume L0 flushes in these cases.
+            Err(CompactionError::Offload(_)) => {}
+            Err(CompactionError::ShuttingDown) => {}
+        };
+
+        result
     }
 
     /// Mutate the timeline with a [`TimelineWriter`].
@@ -2137,6 +2144,13 @@ impl Timeline {
             .unwrap_or(self.conf.default_tenant_conf.checkpoint_timeout)
     }
 
+    fn get_compaction_period(&self) -> Duration {
+        let tenant_conf = self.tenant_conf.load().tenant_conf.clone();
+        tenant_conf
+            .compaction_period
+            .unwrap_or(self.conf.default_tenant_conf.compaction_period)
+    }
+
     fn get_compaction_target_size(&self) -> u64 {
         let tenant_conf = self.tenant_conf.load();
         tenant_conf
@@ -2153,6 +2167,91 @@ impl Timeline {
             .unwrap_or(self.conf.default_tenant_conf.compaction_threshold)
     }
 
+    fn get_l0_flush_delay_threshold(&self) -> Option<usize> {
+        // Default to delay L0 flushes at 3x compaction threshold.
+        const DEFAULT_L0_FLUSH_DELAY_FACTOR: usize = 3;
+
+        // If compaction is disabled, don't delay.
+        if self.get_compaction_period() == Duration::ZERO {
+            return None;
+        }
+
+        let compaction_threshold = self.get_compaction_threshold();
+        let tenant_conf = self.tenant_conf.load();
+        let l0_flush_delay_threshold = tenant_conf
+            .tenant_conf
+            .l0_flush_delay_threshold
+            .or(self.conf.default_tenant_conf.l0_flush_delay_threshold)
+            .unwrap_or(DEFAULT_L0_FLUSH_DELAY_FACTOR * compaction_threshold);
+
+        // 0 disables backpressure.
+        if l0_flush_delay_threshold == 0 {
+            return None;
+        }
+
+        // Clamp the flush delay threshold to the compaction threshold; it doesn't make sense to
+        // backpressure flushes below this.
+        // TODO: the tenant config should have validation to prevent this instead.
+        debug_assert!(l0_flush_delay_threshold >= compaction_threshold);
+        Some(max(l0_flush_delay_threshold, compaction_threshold))
+    }
+
+    fn get_l0_flush_stall_threshold(&self) -> Option<usize> {
+        // Default to stall L0 flushes at 5x compaction threshold.
+        // TODO: stalls are temporarily disabled by default, see below.
+        #[allow(unused)]
+        const DEFAULT_L0_FLUSH_STALL_FACTOR: usize = 5;
+
+        // If compaction is disabled, don't stall.
+        if self.get_compaction_period() == Duration::ZERO {
+            return None;
+        }
+
+        // If compaction is failing, don't stall and try to keep the tenant alive. This may not be a
+        // good idea: read amp can grow unbounded, leading to terrible performance, and we may take
+        // on unbounded compaction debt that can take a long time to fix once compaction comes back
+        // online. At least we'll delay flushes, slowing down the growth and buying some time.
+        if self.compaction_failed.load(AtomicOrdering::Relaxed) {
+            return None;
+        }
+
+        let compaction_threshold = self.get_compaction_threshold();
+        let tenant_conf = self.tenant_conf.load();
+        let l0_flush_stall_threshold = tenant_conf
+            .tenant_conf
+            .l0_flush_stall_threshold
+            .or(self.conf.default_tenant_conf.l0_flush_stall_threshold);
+
+        // Tests sometimes set compaction_threshold=1 to generate lots of layer files, and don't
+        // handle the 20-second compaction delay. Some (e.g. `test_backward_compatibility`) can't
+        // easily adjust the L0 backpressure settings, so just disable stalls in this case.
+        if cfg!(feature = "testing")
+            && compaction_threshold == 1
+            && l0_flush_stall_threshold.is_none()
+        {
+            return None;
+        }
+
+        // Disable stalls by default. In ingest benchmarks, we see image compaction take >10
+        // minutes, blocking L0 compaction, and we can't stall L0 flushes for that long.
+        //
+        // TODO: fix this.
+        // let l0_flush_stall_threshold = l0_flush_stall_threshold
+        //    .unwrap_or(DEFAULT_L0_FLUSH_STALL_FACTOR * compaction_threshold);
+        let l0_flush_stall_threshold = l0_flush_stall_threshold?;
+
+        // 0 disables backpressure.
+        if l0_flush_stall_threshold == 0 {
+            return None;
+        }
+
+        // Clamp the flush stall threshold to the compaction threshold; it doesn't make sense to
+        // backpressure flushes below this.
+        // TODO: the tenant config should have validation to prevent this instead.
+        debug_assert!(l0_flush_stall_threshold >= compaction_threshold);
+        Some(max(l0_flush_stall_threshold, compaction_threshold))
+    }
+
     fn get_image_creation_threshold(&self) -> usize {
         let tenant_conf = self.tenant_conf.load();
         tenant_conf
@@ -2389,6 +2488,7 @@ impl Timeline {
                 gate: Gate::default(),
 
                 compaction_lock: tokio::sync::Mutex::default(),
+                compaction_failed: AtomicBool::default(),
                 gc_lock: tokio::sync::Mutex::default(),
 
                 standby_horizon: AtomicLsn::new(0),
@@ -3604,6 +3704,12 @@ impl Timeline {
         mut layer_flush_start_rx: tokio::sync::watch::Receiver<(u64, Lsn)>,
         ctx: &RequestContext,
     ) {
+        // Subscribe to L0 delta layer updates, for compaction backpressure.
+        let mut watch_l0 = match self.layers.read().await.layer_map() {
+            Ok(lm) => lm.watch_level0_deltas(),
+            Err(Shutdown) => return,
+        };
+
         info!("started flush loop");
         loop {
             tokio::select! {
@@ -3634,43 +3740,62 @@ impl Timeline {
                     break Ok(());
                 }
 
-                let timer = self.metrics.flush_time_histo.start_timer();
-
-                let num_frozen_layers;
-                let frozen_layer_total_size;
-                let layer_to_flush = {
-                    let guard = self.layers.read().await;
-                    let Ok(lm) = guard.layer_map() else {
+                // Fetch the next layer to flush, if any.
+                let (layer, l0_count, frozen_count, frozen_size) = {
+                    let layers = self.layers.read().await;
+                    let Ok(lm) = layers.layer_map() else {
                         info!("dropping out of flush loop for timeline shutdown");
                         return;
                     };
-                    num_frozen_layers = lm.frozen_layers.len();
-                    frozen_layer_total_size = lm
+                    let l0_count = lm.level0_deltas().len();
+                    let frozen_count = lm.frozen_layers.len();
+                    let frozen_size: u64 = lm
                         .frozen_layers
                         .iter()
                         .map(|l| l.estimated_in_mem_size())
-                        .sum::<u64>();
-                    lm.frozen_layers.front().cloned()
-                    // drop 'layers' lock to allow concurrent reads and writes
+                        .sum();
+                    let layer = lm.frozen_layers.front().cloned();
+                    (layer, l0_count, frozen_count, frozen_size)
+                    // drop 'layers' lock
                 };
-                let Some(layer_to_flush) = layer_to_flush else {
+                let Some(layer) = layer else {
                     break Ok(());
                 };
-                if num_frozen_layers
-                    > std::cmp::max(
-                        self.get_compaction_threshold(),
-                        DEFAULT_COMPACTION_THRESHOLD,
-                    )
-                    && frozen_layer_total_size >= /* 128 MB */ 128000000
-                {
-                    tracing::warn!(
-                        "too many frozen layers: {num_frozen_layers} layers with estimated in-mem size of {frozen_layer_total_size} bytes",
-                    );
-                }
-                match self.flush_frozen_layer(layer_to_flush, ctx).await {
-                    Ok(this_layer_to_lsn) => {
-                        flushed_to_lsn = std::cmp::max(flushed_to_lsn, this_layer_to_lsn);
+
+                // Stall flushes to backpressure if compaction can't keep up. This is propagated up
+                // to WAL ingestion by having ephemeral layer rolls wait for flushes.
+                //
+                // NB: the compaction loop only checks `compaction_threshold` every 20 seconds, so
+                // we can end up stalling before compaction even starts. Consider making it more
+                // responsive (e.g. via `watch_level0_deltas`).
+                if let Some(stall_threshold) = self.get_l0_flush_stall_threshold() {
+                    if l0_count >= stall_threshold {
+                        warn!(
+                            "stalling layer flushes for compaction backpressure at {l0_count} \
+                            L0 layers ({frozen_count} frozen layers with {frozen_size} bytes)"
+                        );
+                        let stall_timer = self
+                            .metrics
+                            .flush_delay_histo
+                            .start_timer()
+                            .record_on_drop();
+                        tokio::select! {
+                            result = watch_l0.wait_for(|l0| *l0 < stall_threshold) => {
+                                if let Ok(l0) = result.as_deref() {
+                                    let delay = stall_timer.elapsed().as_secs_f64();
+                                    info!("resuming layer flushes at {l0} L0 layers after {delay:.3}s");
+                                }
+                            },
+                            _ = self.cancel.cancelled() => {},
+                        }
+                        continue; // check again
                     }
+                }
+
+                // Flush the layer.
+                let flush_timer = self.metrics.flush_time_histo.start_timer();
+                match self.flush_frozen_layer(layer, ctx).await {
+                    Ok(layer_lsn) => flushed_to_lsn = max(flushed_to_lsn, layer_lsn),
                     Err(FlushLayerError::Cancelled) => {
                         info!("dropping out of flush loop for timeline shutdown");
                         return;
@@ -3684,7 +3809,30 @@ impl Timeline {
                         break err.map(|_| ());
                     }
                 }
-                timer.stop_and_record();
+                let flush_duration = flush_timer.stop_and_record();
+
+                // Delay the next flush to backpressure if compaction can't keep up. We delay by the
+                // flush duration such that the flush takes 2x as long. This is propagated up to WAL
+                // ingestion by having ephemeral layer rolls wait for flushes.
+                if let Some(delay_threshold) = self.get_l0_flush_delay_threshold() {
+                    if l0_count >= delay_threshold {
+                        let delay = flush_duration.as_secs_f64();
+                        info!(
+                            "delaying layer flush by {delay:.3}s for compaction backpressure at \
+                            {l0_count} L0 layers ({frozen_count} frozen layers with {frozen_size} bytes)"
+                        );
+                        let _delay_timer = self
+                            .metrics
+                            .flush_delay_histo
+                            .start_timer()
+                            .record_on_drop();
+                        tokio::select! {
+                            _ = tokio::time::sleep(flush_duration) => {},
+                            _ = watch_l0.wait_for(|l0| *l0 < delay_threshold) => {},
+                            _ = self.cancel.cancelled() => {},
+                        }
+                    }
+                }
             };
 
             // Unsharded tenants should never advance their LSN beyond the end of the
@@ -3886,24 +4034,6 @@ impl Timeline {
             // release lock on 'layers'
         };
 
-        // Backpressure mechanism: wait with continuation of the flush loop until we have uploaded all layer files.
-        // This makes us refuse ingest until the new layers have been persisted to the remote
-        let start = Instant::now();
-        self.remote_client
-            .wait_completion()
-            .await
-            .map_err(|e| match e {
-                WaitCompletionError::UploadQueueShutDownOrStopped
-                | WaitCompletionError::NotInitialized(
-                    NotInitialized::ShuttingDown | NotInitialized::Stopped,
-                ) => FlushLayerError::Cancelled,
-                WaitCompletionError::NotInitialized(NotInitialized::Uninitialized) => {
-                    FlushLayerError::Other(anyhow!(e).into())
-                }
-            })?;
-        let duration = start.elapsed().as_secs_f64();
-        self.metrics.flush_wait_upload_time_gauge_add(duration);
-
         // FIXME: between create_delta_layer and the scheduling of the upload in `update_metadata_file`,
         // a compaction can delete the file and then it won't be available for uploads any more.
         // We still schedule the upload, resulting in an error, but ideally we'd somehow avoid this
@@ -5932,13 +6062,37 @@ impl TimelineWriter<'_> {
     async fn roll_layer(&mut self, freeze_at: Lsn) -> Result<(), FlushLayerError> {
         let current_size = self.write_guard.as_ref().unwrap().current_size;
 
+        // If layer flushes are backpressured due to compaction not keeping up, wait for the flush
+        // to propagate the backpressure up into WAL ingestion.
+        let l0_count = self
+            .tl
+            .layers
+            .read()
+            .await
+            .layer_map()?
+            .level0_deltas()
+            .len();
+        let wait_thresholds = [
+            self.get_l0_flush_delay_threshold(),
+            self.get_l0_flush_stall_threshold(),
+        ];
+        let wait_threshold = wait_thresholds.into_iter().flatten().min();
+
         // self.write_guard will be taken by the freezing
-        self.tl
+        let flush_id = self
+            .tl
             .freeze_inmem_layer_at(freeze_at, &mut self.write_guard)
             .await?;
 
         assert!(self.write_guard.is_none());
 
+        if let Some(wait_threshold) = wait_threshold {
+            if l0_count >= wait_threshold {
+                info!("layer roll waiting for flush due to compaction backpressure at {l0_count} L0 layers");
+                self.tl.wait_flush_completion(flush_id).await?;
+            }
+        }
+
         if current_size >= self.get_checkpoint_distance() * 2 {
             warn!("Flushed oversized open layer with size {}", current_size)
         }

pageserver/src/tenant/timeline/compaction.rs

@@ -436,12 +436,14 @@ impl KeyHistoryRetention {
         if dry_run {
             return true;
         }
-        let guard = tline.layers.read().await;
-        if !guard.contains_key(key) {
-            return false;
+        let layer_generation;
+        {
+            let guard = tline.layers.read().await;
+            if !guard.contains_key(key) {
+                return false;
+            }
+            layer_generation = guard.get_from_key(key).metadata().generation;
         }
-        let layer_generation = guard.get_from_key(key).metadata().generation;
-        drop(guard);
         if layer_generation == tline.generation {
             info!(
                 key=%key,
@@ -2138,6 +2140,11 @@ impl Timeline {
             self.get_gc_compaction_watermark()
         };
 
+        if compact_below_lsn == Lsn::INVALID {
+            tracing::warn!("no layers to compact with gc: gc_cutoff not generated yet, skipping gc bottom-most compaction");
+            return Ok(vec![]);
+        }
+
         // Split compaction job to about 4GB each
         const GC_COMPACT_MAX_SIZE_MB: u64 = 4 * 1024;
         let sub_compaction_max_job_size_mb =
@@ -2212,6 +2219,12 @@ impl Timeline {
                 } else {
                     end
                 };
+                let end = if ranges_num == idx + 1 {
+                    // extend the compaction range to the end of the key range if it's the last partition
+                    end.max(job.compact_key_range.end)
+                } else {
+                    end
+                };
                 info!(
                     "splitting compaction job: {}..{}, estimated_size={}",
                     start, end, total_size
@@ -2332,6 +2345,11 @@ impl Timeline {
                 // each of the retain_lsn. Therefore, if the user-provided `compact_lsn_range.end` is larger than the real gc cutoff, we will use
                 // the real cutoff.
                 let mut gc_cutoff = if compact_lsn_range.end == Lsn::MAX {
+                    if real_gc_cutoff == Lsn::INVALID {
+                        // If the gc_cutoff is not generated yet, we should not compact anything.
+                        tracing::warn!("no layers to compact with gc: gc_cutoff not generated yet, skipping gc bottom-most compaction");
+                        return Ok(());
+                    }
                     real_gc_cutoff
                 } else {
                     compact_lsn_range.end
@@ -2863,7 +2881,7 @@ impl Timeline {
             "produced {} delta layers and {} image layers, {} layers are kept",
             produced_delta_layers_len,
             produced_image_layers_len,
-            layer_selection.len()
+            keep_layers.len()
         );
 
         // Step 3: Place back to the layer map.
@@ -2909,8 +2927,28 @@ impl Timeline {
         // be batched into `schedule_compaction_update`.
         let disk_consistent_lsn = self.disk_consistent_lsn.load();
         self.schedule_uploads(disk_consistent_lsn, None)?;
+        // If a layer gets rewritten throughout gc-compaction, we need to keep that layer only in `compact_to` instead
+        // of `compact_from`.
+        let compact_from = {
+            let mut compact_from = Vec::new();
+            let mut compact_to_set = HashMap::new();
+            for layer in &compact_to {
+                compact_to_set.insert(layer.layer_desc().key(), layer);
+            }
+            for layer in &layer_selection {
+                if let Some(to) = compact_to_set.get(&layer.layer_desc().key()) {
+                    tracing::info!(
+                        "skipping delete {} because found same layer key at different generation {}",
+                        layer, to
+                    );
+                } else {
+                    compact_from.push(layer.clone());
+                }
+            }
+            compact_from
+        };
         self.remote_client
-            .schedule_compaction_update(&layer_selection, &compact_to)?;
+            .schedule_compaction_update(&compact_from, &compact_to)?;
 
         drop(gc_lock);
 

pageserver/src/tenant/timeline/layer_manager.rs

@@ -337,16 +337,45 @@ impl OpenLayerManager {
         compact_to: &[ResidentLayer],
         metrics: &TimelineMetrics,
     ) {
-        // We can simply reuse compact l0 logic. Use a different function name to indicate a different type of layer map modification.
-        self.finish_compact_l0(compact_from, compact_to, metrics)
+        // gc-compaction could contain layer rewrites. We need to delete the old layers and insert the new ones.
+
+        // Match the old layers with the new layers
+        let mut add_layers = HashMap::new();
+        let mut rewrite_layers = HashMap::new();
+        let mut drop_layers = HashMap::new();
+        for layer in compact_from {
+            drop_layers.insert(layer.layer_desc().key(), layer.clone());
+        }
+        for layer in compact_to {
+            if let Some(old_layer) = drop_layers.remove(&layer.layer_desc().key()) {
+                rewrite_layers.insert(layer.layer_desc().key(), (old_layer.clone(), layer.clone()));
+            } else {
+                add_layers.insert(layer.layer_desc().key(), layer.clone());
+            }
+        }
+        let add_layers = add_layers.values().cloned().collect::<Vec<_>>();
+        let drop_layers = drop_layers.values().cloned().collect::<Vec<_>>();
+        let rewrite_layers = rewrite_layers.values().cloned().collect::<Vec<_>>();
+
+        self.rewrite_layers_inner(&rewrite_layers, &drop_layers, &add_layers, metrics);
     }
 
     /// Called post-compaction when some previous generation image layers were trimmed.
-    pub(crate) fn rewrite_layers(
+    pub fn rewrite_layers(
         &mut self,
         rewrite_layers: &[(Layer, ResidentLayer)],
         drop_layers: &[Layer],
         metrics: &TimelineMetrics,
+    ) {
+        self.rewrite_layers_inner(rewrite_layers, drop_layers, &[], metrics);
+    }
+
+    fn rewrite_layers_inner(
+        &mut self,
+        rewrite_layers: &[(Layer, ResidentLayer)],
+        drop_layers: &[Layer],
+        add_layers: &[ResidentLayer],
+        metrics: &TimelineMetrics,
     ) {
         let mut updates = self.layer_map.batch_update();
         for (old_layer, new_layer) in rewrite_layers {
@@ -382,6 +411,10 @@ impl OpenLayerManager {
         for l in drop_layers {
             Self::delete_historic_layer(l, &mut updates, &mut self.layer_fmgr);
         }
+        for l in add_layers {
+            Self::insert_historic_layer(l.as_ref().clone(), &mut updates, &mut self.layer_fmgr);
+            metrics.record_new_file_metrics(l.layer_desc().file_size);
+        }
         updates.flush();
     }
 

pageserver/src/utilization.rs

@@ -49,7 +49,7 @@ pub(crate) fn regenerate(
     };
 
     // Express a static value for how many shards we may schedule on one node
-    const MAX_SHARDS: u32 = 20000;
+    const MAX_SHARDS: u32 = 5000;
 
     let mut doc = PageserverUtilization {
         disk_usage_bytes: used,

pgxn/neon/neon.c

@@ -19,6 +19,7 @@
 #include "access/xlogrecovery.h"
 #endif
 #include "replication/logical.h"
+#include "replication/logicallauncher.h"
 #include "replication/slot.h"
 #include "replication/walsender.h"
 #include "storage/proc.h"
@@ -434,6 +435,15 @@ _PG_init(void)
 
 	restore_running_xacts_callback = RestoreRunningXactsFromClog;
 
+	DefineCustomBoolVariable(
+							"neon.disable_logical_replication_subscribers",
+							"Disables incomming logical replication",
+							NULL,
+							&disable_logical_replication_subscribers,
+							false,
+							PGC_SIGHUP,
+							0,
+							NULL, NULL, NULL);
 
 	DefineCustomBoolVariable(
 							"neon.allow_replica_misconfig",

poetry.lock

@@ -1,4 +1,4 @@
-# This file is automatically @generated by Poetry 1.8.5 and should not be changed by hand.
+# This file is automatically @generated by Poetry 2.0.1 and should not be changed by hand.
 
 [[package]]
 name = "aiohappyeyeballs"
@@ -6,6 +6,7 @@ version = "2.3.5"
 description = "Happy Eyeballs for asyncio"
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "aiohappyeyeballs-2.3.5-py3-none-any.whl", hash = "sha256:4d6dea59215537dbc746e93e779caea8178c866856a721c9c660d7a5a7b8be03"},
     {file = "aiohappyeyeballs-2.3.5.tar.gz", hash = "sha256:6fa48b9f1317254f122a07a131a86b71ca6946ca989ce6326fff54a99a920105"},
@@ -17,6 +18,7 @@ version = "3.10.11"
 description = "Async http client/server framework (asyncio)"
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "aiohttp-3.10.11-cp310-cp310-macosx_10_9_universal2.whl", hash = "sha256:5077b1a5f40ffa3ba1f40d537d3bec4383988ee51fbba6b74aa8fb1bc466599e"},
     {file = "aiohttp-3.10.11-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:8d6a14a4d93b5b3c2891fca94fa9d41b2322a68194422bef0dd5ec1e57d7d298"},
@@ -128,6 +130,7 @@ version = "1.4.0"
 description = "Postgres integration with asyncio."
 optional = false
 python-versions = ">=3.7"
+groups = ["main"]
 files = [
     {file = "aiopg-1.4.0-py3-none-any.whl", hash = "sha256:aea46e8aff30b039cfa818e6db4752c97656e893fc75e5a5dc57355a9e9dedbd"},
     {file = "aiopg-1.4.0.tar.gz", hash = "sha256:116253bef86b4d954116716d181e9a0294037f266718b2e1c9766af995639d71"},
@@ -146,6 +149,7 @@ version = "1.3.1"
 description = "aiosignal: a list of registered asynchronous callbacks"
 optional = false
 python-versions = ">=3.7"
+groups = ["main"]
 files = [
     {file = "aiosignal-1.3.1-py3-none-any.whl", hash = "sha256:f8376fb07dd1e86a584e4fcdec80b36b7f81aac666ebc724e2c090300dd83b17"},
     {file = "aiosignal-1.3.1.tar.gz", hash = "sha256:54cd96e15e1649b75d6c87526a6ff0b6c1b0dd3459f43d9ca11d48c339b68cfc"},
@@ -160,6 +164,7 @@ version = "2.13.2"
 description = "Allure pytest integration"
 optional = false
 python-versions = "*"
+groups = ["main"]
 files = [
     {file = "allure-pytest-2.13.2.tar.gz", hash = "sha256:22243159e8ec81ce2b5254b4013802198821b1b42f118f69d4a289396607c7b3"},
     {file = "allure_pytest-2.13.2-py3-none-any.whl", hash = "sha256:17de9dbee7f61c8e66a5b5e818b00e419dbcea44cb55c24319401ba813220690"},
@@ -175,6 +180,7 @@ version = "2.13.2"
 description = "Common module for integrate allure with python-based frameworks"
 optional = false
 python-versions = ">=3.6"
+groups = ["main"]
 files = [
     {file = "allure-python-commons-2.13.2.tar.gz", hash = "sha256:8a03681330231b1deadd86b97ff68841c6591320114ae638570f1ed60d7a2033"},
     {file = "allure_python_commons-2.13.2-py3-none-any.whl", hash = "sha256:2bb3646ec3fbf5b36d178a5e735002bc130ae9f9ba80f080af97d368ba375051"},
@@ -190,6 +196,7 @@ version = "0.6.0"
 description = "Reusable constraint types to use with typing.Annotated"
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "annotated_types-0.6.0-py3-none-any.whl", hash = "sha256:0641064de18ba7a25dee8f96403ebc39113d0cb953a01429249d5c7564666a43"},
     {file = "annotated_types-0.6.0.tar.gz", hash = "sha256:563339e807e53ffd9c267e99fc6d9ea23eb8443c08f112651963e24e22f84a5d"},
@@ -201,6 +208,7 @@ version = "4.13.1"
 description = "ANTLR 4.13.1 runtime for Python 3"
 optional = false
 python-versions = "*"
+groups = ["main"]
 files = [
     {file = "antlr4-python3-runtime-4.13.1.tar.gz", hash = "sha256:3cd282f5ea7cfb841537fe01f143350fdb1c0b1ce7981443a2fa8513fddb6d1a"},
     {file = "antlr4_python3_runtime-4.13.1-py3-none-any.whl", hash = "sha256:78ec57aad12c97ac039ca27403ad61cb98aaec8a3f9bb8144f889aa0fa28b943"},
@@ -212,6 +220,7 @@ version = "4.3.0"
 description = "High level compatibility layer for multiple asynchronous event loop implementations"
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "anyio-4.3.0-py3-none-any.whl", hash = "sha256:048e05d0f6caeed70d731f3db756d35dcc1f35747c8c403364a8332c630441b8"},
     {file = "anyio-4.3.0.tar.gz", hash = "sha256:f75253795a87df48568485fd18cdd2a3fa5c4f7c5be8e5e36637733fce06fed6"},
@@ -232,6 +241,7 @@ version = "4.0.3"
 description = "Timeout context manager for asyncio programs"
 optional = false
 python-versions = ">=3.7"
+groups = ["main"]
 files = [
     {file = "async-timeout-4.0.3.tar.gz", hash = "sha256:4640d96be84d82d02ed59ea2b7105a0f7b33abe8703703cd0ab0bf87c427522f"},
     {file = "async_timeout-4.0.3-py3-none-any.whl", hash = "sha256:7405140ff1230c310e51dc27b3145b9092d659ce68ff733fb0cefe3ee42be028"},
@@ -243,6 +253,7 @@ version = "0.30.0"
 description = "An asyncio PostgreSQL driver"
 optional = false
 python-versions = ">=3.8.0"
+groups = ["main"]
 files = [
     {file = "asyncpg-0.30.0-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:bfb4dd5ae0699bad2b233672c8fc5ccbd9ad24b89afded02341786887e37927e"},
     {file = "asyncpg-0.30.0-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:dc1f62c792752a49f88b7e6f774c26077091b44caceb1983509edc18a2222ec0"},
@@ -306,6 +317,7 @@ version = "21.4.0"
 description = "Classes Without Boilerplate"
 optional = false
 python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*"
+groups = ["main"]
 files = [
     {file = "attrs-21.4.0-py2.py3-none-any.whl", hash = "sha256:2d27e3784d7a565d36ab851fe94887c5eccd6a463168875832a1be79c82828b4"},
     {file = "attrs-21.4.0.tar.gz", hash = "sha256:626ba8234211db98e869df76230a137c4c40a12d72445c45d5f5b716f076e2fd"},
@@ -323,6 +335,7 @@ version = "1.88.0"
 description = "AWS SAM Translator is a library that transform SAM templates into AWS CloudFormation templates"
 optional = false
 python-versions = "!=4.0,<=4.0,>=3.8"
+groups = ["main"]
 files = [
     {file = "aws_sam_translator-1.88.0-py3-none-any.whl", hash = "sha256:aa93d498d8de3fb3d485c316155b1628144b823bbc176099a20de06df666fcac"},
     {file = "aws_sam_translator-1.88.0.tar.gz", hash = "sha256:e77c65f3488566122277accd44a0f1ec018e37403e0d5fe25120d96e537e91a7"},
@@ -343,6 +356,7 @@ version = "2.10.0"
 description = "The AWS X-Ray SDK for Python (the SDK) enables Python developers to record and emit information from within their applications to the AWS X-Ray service."
 optional = false
 python-versions = "*"
+groups = ["main"]
 files = [
     {file = "aws-xray-sdk-2.10.0.tar.gz", hash = "sha256:9b14924fd0628cf92936055864655354003f0b1acc3e1c3ffde6403d0799dd7a"},
     {file = "aws_xray_sdk-2.10.0-py2.py3-none-any.whl", hash = "sha256:7551e81a796e1a5471ebe84844c40e8edf7c218db33506d046fec61f7495eda4"},
@@ -358,6 +372,7 @@ version = "2.2.1"
 description = "Function decoration for backoff and retry"
 optional = false
 python-versions = ">=3.7,<4.0"
+groups = ["main"]
 files = [
     {file = "backoff-2.2.1-py3-none-any.whl", hash = "sha256:63579f9a0628e06278f7e47b7d7d5b6ce20dc65c5e96a6f3ca99a6adca0396e8"},
     {file = "backoff-2.2.1.tar.gz", hash = "sha256:03f829f5bb1923180821643f8753b0502c3b682293992485b0eef2807afa5cba"},
@@ -369,6 +384,7 @@ version = "1.34.11"
 description = "The AWS SDK for Python"
 optional = false
 python-versions = ">= 3.8"
+groups = ["main"]
 files = [
     {file = "boto3-1.34.11-py3-none-any.whl", hash = "sha256:1af021e0c6e3040e8de66d403e963566476235bb70f9a8e3f6784813ac2d8026"},
     {file = "boto3-1.34.11.tar.gz", hash = "sha256:31c130a40ec0631059b77d7e87f67ad03ff1685a5b37638ac0c4687026a3259d"},
@@ -388,6 +404,7 @@ version = "1.26.16"
 description = "Type annotations for boto3 1.26.16 generated with mypy-boto3-builder 7.11.11"
 optional = false
 python-versions = ">=3.7"
+groups = ["main"]
 files = [
     {file = "boto3-stubs-1.26.16.tar.gz", hash = "sha256:618253ae19f1480785759bcaee8c8b10ed3fc037027247c26a3461a50f58406d"},
     {file = "boto3_stubs-1.26.16-py3-none-any.whl", hash = "sha256:8cf2925bc3e1349c93eb0f49c1061affc5ca314d69eeb335349037969d0787ed"},
@@ -395,7 +412,6 @@ files = [
 
 [package.dependencies]
 botocore-stubs = "*"
-mypy-boto3-kms = {version = ">=1.26.0,<1.27.0", optional = true, markers = "extra == \"kms\""}
 mypy-boto3-s3 = {version = ">=1.26.0,<1.27.0", optional = true, markers = "extra == \"s3\""}
 types-s3transfer = "*"
 typing-extensions = ">=4.1.0"
@@ -733,6 +749,7 @@ version = "1.34.11"
 description = "Low-level, data-driven core of boto 3."
 optional = false
 python-versions = ">= 3.8"
+groups = ["main"]
 files = [
     {file = "botocore-1.34.11-py3-none-any.whl", hash = "sha256:1ff1398b6ea670e1c01ac67a33af3da854f8e700d3528289c04f319c330d8250"},
     {file = "botocore-1.34.11.tar.gz", hash = "sha256:51905c3d623c60df5dc5794387de7caf886d350180a01a3dfa762e903edb45a9"},
@@ -752,6 +769,7 @@ version = "1.27.38"
 description = "Type annotations for botocore 1.27.38 generated with mypy-boto3-builder 7.10.1"
 optional = false
 python-versions = ">=3.7"
+groups = ["main"]
 files = [
     {file = "botocore-stubs-1.27.38.tar.gz", hash = "sha256:408e8b86b5d171b58f81c74ca9d3b5317a5a8e2d3bc2073aa841ac13b8939e56"},
     {file = "botocore_stubs-1.27.38-py3-none-any.whl", hash = "sha256:7add7641e9a479a9c8366893bb522fd9ca3d58714201e43662a200a148a1bc38"},
@@ -766,6 +784,7 @@ version = "2024.7.4"
 description = "Python package for providing Mozilla's CA Bundle."
 optional = false
 python-versions = ">=3.6"
+groups = ["main"]
 files = [
     {file = "certifi-2024.7.4-py3-none-any.whl", hash = "sha256:c198e21b1289c2ab85ee4e67bb4b4ef3ead0892059901a8d5b622f24a1101e90"},
     {file = "certifi-2024.7.4.tar.gz", hash = "sha256:5a1e7645bc0ec61a09e26c36f6106dd4cf40c6db3a1fb6352b0244e7fb057c7b"},
@@ -777,6 +796,7 @@ version = "1.17.1"
 description = "Foreign Function Interface for Python calling C code."
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "cffi-1.17.1-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:df8b1c11f177bc2313ec4b2d46baec87a5f3e71fc8b45dab2ee7cae86d9aba14"},
     {file = "cffi-1.17.1-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:8f2cdc858323644ab277e9bb925ad72ae0e67f69e804f4898c070998d50b1a67"},
@@ -856,6 +876,7 @@ version = "0.87.1"
 description = "Checks CloudFormation templates for practices and behaviour that could potentially be improved"
 optional = false
 python-versions = "!=4.0,<=4.0,>=3.8"
+groups = ["main"]
 files = [
     {file = "cfn_lint-0.87.1-py3-none-any.whl", hash = "sha256:d450f450635fc223b6f66880ccac52a5fd1a52966fa1705f1ba52b88dfed3071"},
     {file = "cfn_lint-0.87.1.tar.gz", hash = "sha256:b3ce9d3e5e0eadcea5d584c8ccaa00bf2a990a36a64d7ffd8683bc60b7e4f06f"},
@@ -879,6 +900,7 @@ version = "2.1.0"
 description = "The Real First Universal Charset Detector. Open, modern and actively maintained alternative to Chardet."
 optional = false
 python-versions = ">=3.6.0"
+groups = ["main"]
 files = [
     {file = "charset-normalizer-2.1.0.tar.gz", hash = "sha256:575e708016ff3a5e3681541cb9d79312c416835686d054a23accb873b254f413"},
     {file = "charset_normalizer-2.1.0-py3-none-any.whl", hash = "sha256:5189b6f22b01957427f35b6a08d9a0bc45b46d3788ef5a92e978433c7a35f8a5"},
@@ -893,6 +915,7 @@ version = "8.1.3"
 description = "Composable command line interface toolkit"
 optional = false
 python-versions = ">=3.7"
+groups = ["main"]
 files = [
     {file = "click-8.1.3-py3-none-any.whl", hash = "sha256:bb4d8133cb15a609f44e8213d9b391b0809795062913b383c62be0ee95b1db48"},
     {file = "click-8.1.3.tar.gz", hash = "sha256:7682dc8afb30297001674575ea00d1814d808d6a36af415a82bd481d37ba7b8e"},
@@ -907,6 +930,7 @@ version = "0.7.17"
 description = "ClickHouse Database Core Driver for Python, Pandas, and Superset"
 optional = false
 python-versions = "~=3.8"
+groups = ["main"]
 files = [
     {file = "clickhouse-connect-0.7.17.tar.gz", hash = "sha256:854f1f9f3e024e7f89ae5d57cd3289d7a4c3dc91a9f24c4d233014f0ea19cb2d"},
     {file = "clickhouse_connect-0.7.17-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:aca36f5f28be1ada2981fce87724bbf451f267c918015baec59e527de3c9c882"},
@@ -997,6 +1021,8 @@ version = "0.4.5"
 description = "Cross-platform colored terminal text."
 optional = false
 python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*"
+groups = ["main"]
+markers = "sys_platform == \"win32\" or platform_system == \"Windows\""
 files = [
     {file = "colorama-0.4.5-py2.py3-none-any.whl", hash = "sha256:854bf444933e37f5824ae7bfc1e98d5bce2ebe4160d46b5edf346a89358e99da"},
     {file = "colorama-0.4.5.tar.gz", hash = "sha256:e6c6b4334fc50988a639d9b98aa429a0b57da6e17b9a44f0451f930b6967b7a4"},
@@ -1008,6 +1034,7 @@ version = "43.0.1"
 description = "cryptography is a package which provides cryptographic recipes and primitives to Python developers."
 optional = false
 python-versions = ">=3.7"
+groups = ["main"]
 files = [
     {file = "cryptography-43.0.1-cp37-abi3-macosx_10_9_universal2.whl", hash = "sha256:8385d98f6a3bf8bb2d65a73e17ed87a3ba84f6991c155691c51112075f9ffc5d"},
     {file = "cryptography-43.0.1-cp37-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:27e613d7077ac613e399270253259d9d53872aaf657471473ebfc9a52935c062"},
@@ -1057,6 +1084,7 @@ version = "7.1.0"
 description = "A Python library for the Docker Engine API."
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "docker-7.1.0-py3-none-any.whl", hash = "sha256:c96b93b7f0a746f9e77d325bcfb87422a3d8bd4f03136ae8a85b37f1898d5fc0"},
     {file = "docker-7.1.0.tar.gz", hash = "sha256:ad8c70e6e3f8926cb8a92619b832b4ea5299e2831c14284663184e200546fa6c"},
@@ -1079,6 +1107,7 @@ version = "1.9.0"
 description = "execnet: rapid multi-Python deployment"
 optional = false
 python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*"
+groups = ["main"]
 files = [
     {file = "execnet-1.9.0-py2.py3-none-any.whl", hash = "sha256:a295f7cc774947aac58dde7fdc85f4aa00c42adf5d8f5468fc630c1acf30a142"},
     {file = "execnet-1.9.0.tar.gz", hash = "sha256:8f694f3ba9cc92cab508b152dcfe322153975c29bda272e2fd7f3f00f36e47c5"},
@@ -1093,6 +1122,7 @@ version = "2.2.5"
 description = "A simple framework for building complex web applications."
 optional = false
 python-versions = ">=3.7"
+groups = ["main"]
 files = [
     {file = "Flask-2.2.5-py3-none-any.whl", hash = "sha256:58107ed83443e86067e41eff4631b058178191a355886f8e479e347fa1285fdf"},
     {file = "Flask-2.2.5.tar.gz", hash = "sha256:edee9b0a7ff26621bd5a8c10ff484ae28737a2410d99b0bb9a6850c7fb977aa0"},
@@ -1114,6 +1144,7 @@ version = "5.0.0"
 description = "A Flask extension adding a decorator for CORS support"
 optional = false
 python-versions = "*"
+groups = ["main"]
 files = [
     {file = "Flask_Cors-5.0.0-py2.py3-none-any.whl", hash = "sha256:b9e307d082a9261c100d8fb0ba909eec6a228ed1b60a8315fd85f783d61910bc"},
     {file = "flask_cors-5.0.0.tar.gz", hash = "sha256:5aadb4b950c4e93745034594d9f3ea6591f734bb3662e16e255ffbf5e89c88ef"},
@@ -1128,6 +1159,7 @@ version = "1.5.0"
 description = "A list-like structure which implements collections.abc.MutableSequence"
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "frozenlist-1.5.0-cp310-cp310-macosx_10_9_universal2.whl", hash = "sha256:5b6a66c18b5b9dd261ca98dffcb826a525334b2f29e7caa54e182255c5f6a65a"},
     {file = "frozenlist-1.5.0-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:d1b3eb7b05ea246510b43a7e53ed1653e55c2121019a97e60cad7efb881a97bb"},
@@ -1229,6 +1261,7 @@ version = "3.2.1"
 description = "GraphQL implementation for Python, a port of GraphQL.js, the JavaScript reference implementation for GraphQL."
 optional = false
 python-versions = ">=3.6,<4"
+groups = ["main"]
 files = [
     {file = "graphql-core-3.2.1.tar.gz", hash = "sha256:9d1bf141427b7d54be944587c8349df791ce60ade2e3cccaf9c56368c133c201"},
     {file = "graphql_core-3.2.1-py3-none-any.whl", hash = "sha256:f83c658e4968998eed1923a2e3e3eddd347e005ac0315fbb7ca4d70ea9156323"},
@@ -1240,6 +1273,7 @@ version = "0.14.0"
 description = "A pure-Python, bring-your-own-I/O implementation of HTTP/1.1"
 optional = false
 python-versions = ">=3.7"
+groups = ["main"]
 files = [
     {file = "h11-0.14.0-py3-none-any.whl", hash = "sha256:e3fe4ac4b851c468cc8363d500db52c2ead036020723024a109d37346efaa761"},
     {file = "h11-0.14.0.tar.gz", hash = "sha256:8f19fbbe99e72420ff35c00b27a34cb9937e902a8b810e2c88300c6f0a3b699d"},
@@ -1248,27 +1282,33 @@ files = [
 [[package]]
 name = "h2"
 version = "4.1.0"
-description = "HTTP/2 State-Machine based protocol implementation"
+description = "Pure-Python HTTP/2 protocol implementation"
 optional = false
-python-versions = ">=3.6.1"
-files = [
-    {file = "h2-4.1.0-py3-none-any.whl", hash = "sha256:03a46bcf682256c95b5fd9e9a99c1323584c3eec6440d379b9903d709476bc6d"},
-    {file = "h2-4.1.0.tar.gz", hash = "sha256:a83aca08fbe7aacb79fec788c9c0bac936343560ed9ec18b82a13a12c28d2abb"},
-]
+python-versions = ">=3.9"
+groups = ["main"]
+files = []
+develop = false
 
 [package.dependencies]
-hpack = ">=4.0,<5"
-hyperframe = ">=6.0,<7"
+hpack = ">=4.1,<5"
+hyperframe = ">=6.1,<7"
+
+[package.source]
+type = "git"
+url = "https://github.com/python-hyper/h2"
+reference = "HEAD"
+resolved_reference = "0b98b244b5fd1fe96100ac14905417a3b70a4286"
 
 [[package]]
 name = "hpack"
-version = "4.0.0"
-description = "Pure-Python HPACK header compression"
+version = "4.1.0"
+description = "Pure-Python HPACK header encoding"
 optional = false
-python-versions = ">=3.6.1"
+python-versions = ">=3.9"
+groups = ["main"]
 files = [
-    {file = "hpack-4.0.0-py3-none-any.whl", hash = "sha256:84a076fad3dc9a9f8063ccb8041ef100867b1878b25ef0ee63847a5d53818a6c"},
-    {file = "hpack-4.0.0.tar.gz", hash = "sha256:fc41de0c63e687ebffde81187a948221294896f6bdc0ae2312708df339430095"},
+    {file = "hpack-4.1.0-py3-none-any.whl", hash = "sha256:157ac792668d995c657d93111f46b4535ed114f0c9c8d672271bbec7eae1b496"},
+    {file = "hpack-4.1.0.tar.gz", hash = "sha256:ec5eca154f7056aa06f196a557655c5b009b382873ac8d1e66e79e87535f1dca"},
 ]
 
 [[package]]
@@ -1277,6 +1317,7 @@ version = "1.0.3"
 description = "A minimal low-level HTTP client."
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "httpcore-1.0.3-py3-none-any.whl", hash = "sha256:9a6a501c3099307d9fd76ac244e08503427679b1e81ceb1d922485e2f2462ad2"},
     {file = "httpcore-1.0.3.tar.gz", hash = "sha256:5c0f9546ad17dac4d0772b0808856eb616eb8b48ce94f49ed819fd6982a8a544"},
@@ -1298,6 +1339,7 @@ version = "0.26.0"
 description = "The next generation HTTP client."
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "httpx-0.26.0-py3-none-any.whl", hash = "sha256:8915f5a3627c4d47b73e8202457cb28f1266982d1159bd5779d86a80c0eab1cd"},
     {file = "httpx-0.26.0.tar.gz", hash = "sha256:451b55c30d5185ea6b23c2c793abf9bb237d2a7dfb901ced6ff69ad37ec1dfaf"},
@@ -1319,13 +1361,14 @@ socks = ["socksio (==1.*)"]
 
 [[package]]
 name = "hyperframe"
-version = "6.0.1"
-description = "HTTP/2 framing layer for Python"
+version = "6.1.0"
+description = "Pure-Python HTTP/2 framing"
 optional = false
-python-versions = ">=3.6.1"
+python-versions = ">=3.9"
+groups = ["main"]
 files = [
-    {file = "hyperframe-6.0.1-py3-none-any.whl", hash = "sha256:0ec6bafd80d8ad2195c4f03aacba3a8265e57bc4cff261e802bf39970ed02a15"},
-    {file = "hyperframe-6.0.1.tar.gz", hash = "sha256:ae510046231dc8e9ecb1a6586f63d2347bf4c8905914aa84ba585ae85f28a914"},
+    {file = "hyperframe-6.1.0-py3-none-any.whl", hash = "sha256:b03380493a519fce58ea5af42e4a42317bf9bd425596f7a0835ffce80f1a42e5"},
+    {file = "hyperframe-6.1.0.tar.gz", hash = "sha256:f630908a00854a7adeabd6382b43923a4c4cd4b821fcb527e6ab9e15382a3b08"},
 ]
 
 [[package]]
@@ -1334,6 +1377,7 @@ version = "3.7"
 description = "Internationalized Domain Names in Applications (IDNA)"
 optional = false
 python-versions = ">=3.5"
+groups = ["main"]
 files = [
     {file = "idna-3.7-py3-none-any.whl", hash = "sha256:82fee1fc78add43492d3a1898bfa6d8a904cc97d8427f683ed8e798d07761aa0"},
     {file = "idna-3.7.tar.gz", hash = "sha256:028ff3aadf0609c1fd278d8ea3089299412a7a8b9bd005dd08b9f8285bcb5cfc"},
@@ -1345,6 +1389,7 @@ version = "1.1.1"
 description = "iniconfig: brain-dead simple config-ini parsing"
 optional = false
 python-versions = "*"
+groups = ["main"]
 files = [
     {file = "iniconfig-1.1.1-py2.py3-none-any.whl", hash = "sha256:011e24c64b7f47f6ebd835bb12a743f2fbe9a26d4cecaa7f53bc4f35ee9da8b3"},
     {file = "iniconfig-1.1.1.tar.gz", hash = "sha256:bc3af051d7d14b2ee5ef9969666def0cd1a000e121eaea580d4a313df4b37f32"},
@@ -1356,6 +1401,7 @@ version = "2.1.2"
 description = "Safely pass data to untrusted environments and back."
 optional = false
 python-versions = ">=3.7"
+groups = ["main"]
 files = [
     {file = "itsdangerous-2.1.2-py3-none-any.whl", hash = "sha256:2c2349112351b88699d8d4b6b075022c0808887cb7ad10069318a8b0bc88db44"},
     {file = "itsdangerous-2.1.2.tar.gz", hash = "sha256:5dbbc68b317e5e42f327f9021763545dc3fc3bfe22e6deb96aaf1fc38874156a"},
@@ -1367,6 +1413,7 @@ version = "3.1.5"
 description = "A very fast and expressive template engine."
 optional = false
 python-versions = ">=3.7"
+groups = ["main"]
 files = [
     {file = "jinja2-3.1.5-py3-none-any.whl", hash = "sha256:aba0f4dc9ed8013c424088f68a5c226f7d6097ed89b246d7749c2ec4175c6adb"},
     {file = "jinja2-3.1.5.tar.gz", hash = "sha256:8fefff8dc3034e27bb80d67c671eb8a9bc424c0ef4c0826edbff304cceff43bb"},
@@ -1384,6 +1431,7 @@ version = "1.0.1"
 description = "JSON Matching Expressions"
 optional = false
 python-versions = ">=3.7"
+groups = ["main"]
 files = [
     {file = "jmespath-1.0.1-py3-none-any.whl", hash = "sha256:02e2e4cc71b5bcab88332eebf907519190dd9e6e82107fa7f83b1003a6252980"},
     {file = "jmespath-1.0.1.tar.gz", hash = "sha256:90261b206d6defd58fdd5e85f478bf633a2901798906be2ad389150c5c60edbe"},
@@ -1395,6 +1443,7 @@ version = "0.9.0"
 description = "The ultimate Python library for JOSE RFCs, including JWS, JWE, JWK, JWA, JWT"
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "joserfc-0.9.0-py3-none-any.whl", hash = "sha256:4026bdbe2c196cd40574e916fa1e28874d99649412edaab0e373dec3077153fb"},
     {file = "joserfc-0.9.0.tar.gz", hash = "sha256:eebca7f587b1761ce43a98ffd5327f2b600b9aa5bb0a77b947687f503ad43bc0"},
@@ -1412,6 +1461,7 @@ version = "1.2.3"
 description = "Generate source code for Python classes from a JSON schema."
 optional = false
 python-versions = ">= 2.7"
+groups = ["main"]
 files = [
     {file = "jschema_to_python-1.2.3-py3-none-any.whl", hash = "sha256:8a703ca7604d42d74b2815eecf99a33359a8dccbb80806cce386d5e2dd992b05"},
     {file = "jschema_to_python-1.2.3.tar.gz", hash = "sha256:76ff14fe5d304708ccad1284e4b11f96a658949a31ee7faed9e0995279549b91"},
@@ -1428,6 +1478,7 @@ version = "2.0.0"
 description = "Diff JSON and JSON-like structures in Python"
 optional = false
 python-versions = "*"
+groups = ["main"]
 files = [
     {file = "jsondiff-2.0.0-py3-none-any.whl", hash = "sha256:689841d66273fc88fc79f7d33f4c074774f4f214b6466e3aff0e5adaf889d1e0"},
     {file = "jsondiff-2.0.0.tar.gz", hash = "sha256:2795844ef075ec8a2b8d385c4d59f5ea48b08e7180fce3cb2787be0db00b1fb4"},
@@ -1439,6 +1490,8 @@ version = "0.20.0"
 description = "Python bindings for Jsonnet - The data templating language"
 optional = false
 python-versions = "*"
+groups = ["main"]
+markers = "python_version < \"3.13\""
 files = [
     {file = "jsonnet-0.20.0.tar.gz", hash = "sha256:7e770c7bf3a366b97b650a39430450f77612e74406731eb75c5bd59f3f104d4f"},
 ]
@@ -1449,6 +1502,7 @@ version = "1.32"
 description = "Apply JSON-Patches (RFC 6902)"
 optional = false
 python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*"
+groups = ["main"]
 files = [
     {file = "jsonpatch-1.32-py2.py3-none-any.whl", hash = "sha256:26ac385719ac9f54df8a2f0827bb8253aa3ea8ab7b3368457bcdb8c14595a397"},
     {file = "jsonpatch-1.32.tar.gz", hash = "sha256:b6ddfe6c3db30d81a96aaeceb6baf916094ffa23d7dd5fa2c13e13f8b6e600c2"},
@@ -1463,6 +1517,7 @@ version = "1.6.1"
 description = "A final implementation of JSONPath for Python that aims to be standard compliant, including arithmetic and binary comparison operators and providing clear AST for metaprogramming."
 optional = false
 python-versions = "*"
+groups = ["main"]
 files = [
     {file = "jsonpath-ng-1.6.1.tar.gz", hash = "sha256:086c37ba4917304850bd837aeab806670224d3f038fe2833ff593a672ef0a5fa"},
     {file = "jsonpath_ng-1.6.1-py3-none-any.whl", hash = "sha256:8f22cd8273d7772eea9aaa84d922e0841aa36fdb8a2c6b7f6c3791a16a9bc0be"},
@@ -1477,6 +1532,7 @@ version = "2.2.0"
 description = "Python library for serializing any arbitrary object graph into JSON"
 optional = false
 python-versions = ">=2.7"
+groups = ["main"]
 files = [
     {file = "jsonpickle-2.2.0-py2.py3-none-any.whl", hash = "sha256:de7f2613818aa4f234138ca11243d6359ff83ae528b2185efdd474f62bcf9ae1"},
     {file = "jsonpickle-2.2.0.tar.gz", hash = "sha256:7b272918b0554182e53dc340ddd62d9b7f902fec7e7b05620c04f3ccef479a0e"},
@@ -1493,6 +1549,7 @@ version = "2.3"
 description = "Identify specific nodes in a JSON document (RFC 6901)"
 optional = false
 python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*"
+groups = ["main"]
 files = [
     {file = "jsonpointer-2.3-py2.py3-none-any.whl", hash = "sha256:51801e558539b4e9cd268638c078c6c5746c9ac96bc38152d443400e4f3793e9"},
     {file = "jsonpointer-2.3.tar.gz", hash = "sha256:97cba51526c829282218feb99dab1b1e6bdf8efd1c43dc9d57be093c0d69c99a"},
@@ -1504,6 +1561,7 @@ version = "4.17.3"
 description = "An implementation of JSON Schema validation for Python"
 optional = false
 python-versions = ">=3.7"
+groups = ["main"]
 files = [
     {file = "jsonschema-4.17.3-py3-none-any.whl", hash = "sha256:a870ad254da1a8ca84b6a2905cac29d265f805acc57af304784962a2aa6508f6"},
     {file = "jsonschema-4.17.3.tar.gz", hash = "sha256:0f864437ab8b6076ba6707453ef8f98a6a0d512a80e93f8abdb676f737ecb60d"},
@@ -1523,6 +1581,7 @@ version = "0.1.6"
 description = "JSONSchema Spec with object-oriented paths"
 optional = false
 python-versions = ">=3.7.0,<4.0.0"
+groups = ["main"]
 files = [
     {file = "jsonschema_spec-0.1.6-py3-none-any.whl", hash = "sha256:f2206d18c89d1824c1f775ba14ed039743b41a9167bd2c5bdb774b66b3ca0bbf"},
     {file = "jsonschema_spec-0.1.6.tar.gz", hash = "sha256:90215863b56e212086641956b20127ccbf6d8a3a38343dad01d6a74d19482f76"},
@@ -1540,6 +1599,7 @@ version = "1.9"
 description = "Creates JUnit XML test result documents that can be read by tools such as Jenkins"
 optional = false
 python-versions = "*"
+groups = ["main"]
 files = [
     {file = "junit-xml-1.9.tar.gz", hash = "sha256:de16a051990d4e25a3982b2dd9e89d671067548718866416faec14d9de56db9f"},
     {file = "junit_xml-1.9-py2.py3-none-any.whl", hash = "sha256:ec5ca1a55aefdd76d28fcc0b135251d156c7106fa979686a4b48d62b761b4732"},
@@ -1554,6 +1614,7 @@ version = "1.5.6"
 description = "Implementation of JOSE Web standards"
 optional = false
 python-versions = ">= 3.8"
+groups = ["main"]
 files = [
     {file = "jwcrypto-1.5.6-py3-none-any.whl", hash = "sha256:150d2b0ebbdb8f40b77f543fb44ffd2baeff48788be71f67f03566692fd55789"},
     {file = "jwcrypto-1.5.6.tar.gz", hash = "sha256:771a87762a0c081ae6166958a954f80848820b2ab066937dc8b8379d65b1b039"},
@@ -1569,6 +1630,7 @@ version = "2.0.2"
 description = "Pure Python client for Apache Kafka"
 optional = false
 python-versions = "*"
+groups = ["main"]
 files = [
     {file = "kafka-python-2.0.2.tar.gz", hash = "sha256:04dfe7fea2b63726cd6f3e79a2d86e709d608d74406638c5da33a01d45a9d7e3"},
     {file = "kafka_python-2.0.2-py2.py3-none-any.whl", hash = "sha256:2d92418c7cb1c298fa6c7f0fb3519b520d0d7526ac6cb7ae2a4fc65a51a94b6e"},
@@ -1583,6 +1645,7 @@ version = "1.10.0"
 description = "A fast and thorough lazy object proxy."
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "lazy-object-proxy-1.10.0.tar.gz", hash = "sha256:78247b6d45f43a52ef35c25b5581459e85117225408a4128a3daf8bf9648ac69"},
     {file = "lazy_object_proxy-1.10.0-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:855e068b0358ab916454464a884779c7ffa312b8925c6f7401e952dcf3b89977"},
@@ -1629,6 +1692,7 @@ version = "4.3.3"
 description = "LZ4 Bindings for Python"
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "lz4-4.3.3-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:b891880c187e96339474af2a3b2bfb11a8e4732ff5034be919aa9029484cd201"},
     {file = "lz4-4.3.3-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:222a7e35137d7539c9c33bb53fcbb26510c5748779364014235afc62b0ec797f"},
@@ -1679,6 +1743,7 @@ version = "2.1.1"
 description = "Safely add untrusted strings to HTML/XML markup."
 optional = false
 python-versions = ">=3.7"
+groups = ["main"]
 files = [
     {file = "MarkupSafe-2.1.1-cp310-cp310-macosx_10_9_universal2.whl", hash = "sha256:86b1f75c4e7c2ac2ccdaec2b9022845dbb81880ca318bb7a0a01fbf7813e3812"},
     {file = "MarkupSafe-2.1.1-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:f121a1420d4e173a5d96e47e9a0c0dcff965afdf1626d28de1460815f7c4ee7a"},
@@ -1728,6 +1793,7 @@ version = "5.0.6"
 description = ""
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "moto-5.0.6-py2.py3-none-any.whl", hash = "sha256:ca1e22831a741733b581ff2ef4d6ae2e1c6db1eab97af1b78b86ca2c6e88c609"},
     {file = "moto-5.0.6.tar.gz", hash = "sha256:ad8b23f2b555ad694da8b2432a42b6d96beaaf67a4e7d932196a72193a2eee2c"},
@@ -1787,6 +1853,7 @@ version = "1.3.0"
 description = "Python library for arbitrary-precision floating-point arithmetic"
 optional = false
 python-versions = "*"
+groups = ["main"]
 files = [
     {file = "mpmath-1.3.0-py3-none-any.whl", hash = "sha256:a0b2b9fe80bbcd81a6647ff13108738cfb482d481d826cc0e02f5b35e5c88d2c"},
     {file = "mpmath-1.3.0.tar.gz", hash = "sha256:7a28eb2a9774d00c7bc92411c19a89209d5da7c4c9a9e227be8330a23a25b91f"},
@@ -1804,6 +1871,7 @@ version = "6.0.5"
 description = "multidict implementation"
 optional = false
 python-versions = ">=3.7"
+groups = ["main"]
 files = [
     {file = "multidict-6.0.5-cp310-cp310-macosx_10_9_universal2.whl", hash = "sha256:228b644ae063c10e7f324ab1ab6b548bdf6f8b47f3ec234fef1093bc2735e5f9"},
     {file = "multidict-6.0.5-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:896ebdcf62683551312c30e20614305f53125750803b614e9e6ce74a96232604"},
@@ -1903,6 +1971,7 @@ version = "1.13.0"
 description = "Optional static typing for Python"
 optional = false
 python-versions = ">=3.8"
+groups = ["dev"]
 files = [
     {file = "mypy-1.13.0-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:6607e0f1dd1fb7f0aca14d936d13fd19eba5e17e1cd2a14f808fa5f8f6d8f60a"},
     {file = "mypy-1.13.0-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:8a21be69bd26fa81b1f80a61ee7ab05b076c674d9b18fb56239d72e21d9f4c80"},
@@ -1949,23 +2018,13 @@ install-types = ["pip"]
 mypyc = ["setuptools (>=50)"]
 reports = ["lxml"]
 
-[[package]]
-name = "mypy-boto3-kms"
-version = "1.26.147"
-description = "Type annotations for boto3.KMS 1.26.147 service generated with mypy-boto3-builder 7.14.5"
-optional = false
-python-versions = ">=3.7"
-files = [
-    {file = "mypy-boto3-kms-1.26.147.tar.gz", hash = "sha256:816a4d1bb0585e1b9620a3f96c1d69a06f53b7b5621858579dd77c60dbb5fa5c"},
-    {file = "mypy_boto3_kms-1.26.147-py3-none-any.whl", hash = "sha256:493f0db674a25c88769f5cb8ab8ac00d3dda5dfc903d5cda34c990ee64689f79"},
-]
-
 [[package]]
 name = "mypy-boto3-s3"
 version = "1.26.0.post1"
 description = "Type annotations for boto3.S3 1.26.0 service generated with mypy-boto3-builder 7.11.10"
 optional = false
 python-versions = ">=3.7"
+groups = ["main"]
 files = [
     {file = "mypy-boto3-s3-1.26.0.post1.tar.gz", hash = "sha256:6d7079f8c739dc993cbedad0736299c413b297814b73795a3855a79169ecc938"},
     {file = "mypy_boto3_s3-1.26.0.post1-py3-none-any.whl", hash = "sha256:7de2792ff0cc541b84cd46ff3a6aa2b6e5f267217f2203f27f6e4016bddc644d"},
@@ -1980,6 +2039,7 @@ version = "1.0.0"
 description = "Type system extensions for programs checked with the mypy type checker."
 optional = false
 python-versions = ">=3.5"
+groups = ["dev"]
 files = [
     {file = "mypy_extensions-1.0.0-py3-none-any.whl", hash = "sha256:4392f6c0eb8a5668a69e23d168ffa70f0be9ccfd32b5cc2d26a34ae5b844552d"},
     {file = "mypy_extensions-1.0.0.tar.gz", hash = "sha256:75dbf8955dc00442a438fc4d0666508a9a97b6bd41aa2f0ffe9d2f2725af0782"},
@@ -1991,6 +2051,7 @@ version = "2.8.5"
 description = "Python package for creating and manipulating graphs and networks"
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "networkx-2.8.5-py3-none-any.whl", hash = "sha256:a762f4b385692d9c3a6f2912d058d76d29a827deaedf9e63ed14d397b8030687"},
     {file = "networkx-2.8.5.tar.gz", hash = "sha256:15a7b81a360791c458c55a417418ea136c13378cfdc06a2dcdc12bd2f9cf09c1"},
@@ -2009,6 +2070,7 @@ version = "0.4.4"
 description = "OpenAPI schema validation for Python"
 optional = false
 python-versions = ">=3.7.0,<4.0.0"
+groups = ["main"]
 files = [
     {file = "openapi_schema_validator-0.4.4-py3-none-any.whl", hash = "sha256:79f37f38ef9fd5206b924ed7a6f382cea7b649b3b56383c47f1906082b7b9015"},
     {file = "openapi_schema_validator-0.4.4.tar.gz", hash = "sha256:c573e2be2c783abae56c5a1486ab716ca96e09d1c3eab56020d1dc680aa57bf8"},
@@ -2027,6 +2089,7 @@ version = "0.5.7"
 description = "OpenAPI 2.0 (aka Swagger) and OpenAPI 3 spec validator"
 optional = false
 python-versions = ">=3.7.0,<4.0.0"
+groups = ["main"]
 files = [
     {file = "openapi_spec_validator-0.5.7-py3-none-any.whl", hash = "sha256:8712d2879db7692974ef89c47a3ebfc79436442921ec3a826ac0ce80cde8c549"},
     {file = "openapi_spec_validator-0.5.7.tar.gz", hash = "sha256:6c2d42180045a80fd6314de848b94310bdb0fa4949f4b099578b69f79d9fa5ac"},
@@ -2044,6 +2107,7 @@ version = "24.2"
 description = "Core utilities for Python packages"
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "packaging-24.2-py3-none-any.whl", hash = "sha256:09abb1bccd265c01f4a3aa3f7a7db064b36514d2cba19a2f694fe6150451a759"},
     {file = "packaging-24.2.tar.gz", hash = "sha256:c228a6dc5e932d346bc5739379109d49e8853dd8223571c7c5b55260edc0b97f"},
@@ -2055,6 +2119,7 @@ version = "0.4.3"
 description = "Object-oriented paths"
 optional = false
 python-versions = ">=3.7.0,<4.0.0"
+groups = ["main"]
 files = [
     {file = "pathable-0.4.3-py3-none-any.whl", hash = "sha256:cdd7b1f9d7d5c8b8d3315dbf5a86b2596053ae845f056f57d97c0eefff84da14"},
     {file = "pathable-0.4.3.tar.gz", hash = "sha256:5c869d315be50776cc8a993f3af43e0c60dc01506b399643f919034ebf4cdcab"},
@@ -2066,6 +2131,7 @@ version = "5.9.0"
 description = "Python Build Reasonableness"
 optional = false
 python-versions = ">=2.6"
+groups = ["main"]
 files = [
     {file = "pbr-5.9.0-py2.py3-none-any.whl", hash = "sha256:e547125940bcc052856ded43be8e101f63828c2d94239ffbe2b327ba3d5ccf0a"},
     {file = "pbr-5.9.0.tar.gz", hash = "sha256:e8dca2f4b43560edef58813969f52a56cef023146cbb8931626db80e6c1c4308"},
@@ -2077,6 +2143,7 @@ version = "1.0.0"
 description = "plugin and hook calling mechanisms for python"
 optional = false
 python-versions = ">=3.6"
+groups = ["main"]
 files = [
     {file = "pluggy-1.0.0-py2.py3-none-any.whl", hash = "sha256:74134bbf457f031a36d68416e1509f34bd5ccc019f0bcc952c7b909d06b37bd3"},
     {file = "pluggy-1.0.0.tar.gz", hash = "sha256:4224373bacce55f955a878bf9cfa763c1e360858e330072059e10bad68531159"},
@@ -2092,6 +2159,7 @@ version = "3.11"
 description = "Python Lex & Yacc"
 optional = false
 python-versions = "*"
+groups = ["main"]
 files = [
     {file = "ply-3.11-py2.py3-none-any.whl", hash = "sha256:096f9b8350b65ebd2fd1346b12452efe5b9607f7482813ffca50c22722a807ce"},
     {file = "ply-3.11.tar.gz", hash = "sha256:00c7c1aaa88358b9c765b6d3000c6eec0ba42abca5351b095321aef446081da3"},
@@ -2103,6 +2171,7 @@ version = "0.14.1"
 description = "Python client for the Prometheus monitoring system."
 optional = false
 python-versions = ">=3.6"
+groups = ["main"]
 files = [
     {file = "prometheus_client-0.14.1-py3-none-any.whl", hash = "sha256:522fded625282822a89e2773452f42df14b5a8e84a86433e3f8a189c1d54dc01"},
     {file = "prometheus_client-0.14.1.tar.gz", hash = "sha256:5459c427624961076277fdc6dc50540e2bacb98eebde99886e59ec55ed92093a"},
@@ -2117,6 +2186,7 @@ version = "0.2.0"
 description = "Accelerated property cache"
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "propcache-0.2.0-cp310-cp310-macosx_10_9_universal2.whl", hash = "sha256:c5869b8fd70b81835a6f187c5fdbe67917a04d7e52b6e7cc4e5fe39d55c39d58"},
     {file = "propcache-0.2.0-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:952e0d9d07609d9c5be361f33b0d6d650cd2bae393aabb11d9b719364521984b"},
@@ -2224,6 +2294,7 @@ version = "5.9.4"
 description = "Cross-platform lib for process and system monitoring in Python."
 optional = false
 python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*"
+groups = ["main"]
 files = [
     {file = "psutil-5.9.4-cp27-cp27m-macosx_10_9_x86_64.whl", hash = "sha256:c1ca331af862803a42677c120aff8a814a804e09832f166f226bfd22b56feee8"},
     {file = "psutil-5.9.4-cp27-cp27m-manylinux2010_i686.whl", hash = "sha256:68908971daf802203f3d37e78d3f8831b6d1014864d7a85937941bb35f09aefe"},
@@ -2250,6 +2321,7 @@ version = "2.9.10"
 description = "psycopg2 - Python-PostgreSQL Database Adapter"
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "psycopg2-binary-2.9.10.tar.gz", hash = "sha256:4b3df0e6990aa98acda57d983942eff13d824135fe2250e6522edaa782a06de2"},
     {file = "psycopg2_binary-2.9.10-cp310-cp310-macosx_12_0_x86_64.whl", hash = "sha256:0ea8e3d0ae83564f2fc554955d327fa081d065c8ca5cc6d2abb643e2c9c1200f"},
@@ -2327,6 +2399,7 @@ version = "0.5.4"
 description = "Pure Python PartiQL Parser"
 optional = false
 python-versions = "*"
+groups = ["main"]
 files = [
     {file = "py_partiql_parser-0.5.4-py2.py3-none-any.whl", hash = "sha256:3dc4295a47da9587681a96b35c6e151886fdbd0a4acbe0d97c4c68e5f689d315"},
     {file = "py_partiql_parser-0.5.4.tar.gz", hash = "sha256:72e043919538fa63edae72fb59afc7e3fd93adbde656718a7d2b4666f23dd114"},
@@ -2341,6 +2414,7 @@ version = "2.21"
 description = "C parser in Python"
 optional = false
 python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*"
+groups = ["main"]
 files = [
     {file = "pycparser-2.21-py2.py3-none-any.whl", hash = "sha256:8ee45429555515e1f6b185e78100aea234072576aa43ab53aefcae078162fca9"},
     {file = "pycparser-2.21.tar.gz", hash = "sha256:e644fdec12f7872f86c58ff790da456218b10f863970249516d60a5eaca77206"},
@@ -2352,6 +2426,7 @@ version = "2.10.4"
 description = "Data validation using Python type hints"
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "pydantic-2.10.4-py3-none-any.whl", hash = "sha256:597e135ea68be3a37552fb524bc7d0d66dcf93d395acd93a00682f1efcb8ee3d"},
     {file = "pydantic-2.10.4.tar.gz", hash = "sha256:82f12e9723da6de4fe2ba888b5971157b3be7ad914267dea8f05f82b28254f06"},
@@ -2372,6 +2447,7 @@ version = "2.27.2"
 description = "Core functionality for Pydantic validation and serialization"
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "pydantic_core-2.27.2-cp310-cp310-macosx_10_12_x86_64.whl", hash = "sha256:2d367ca20b2f14095a8f4fa1210f5a7b78b8a20009ecced6b12818f455b1e9fa"},
     {file = "pydantic_core-2.27.2-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:491a2b73db93fab69731eaee494f320faa4e093dbed776be1a829c2eb222c34c"},
@@ -2484,6 +2560,7 @@ version = "2.4.0"
 description = "JSON Web Token implementation in Python"
 optional = false
 python-versions = ">=3.6"
+groups = ["main"]
 files = [
     {file = "PyJWT-2.4.0-py3-none-any.whl", hash = "sha256:72d1d253f32dbd4f5c88eaf1fdc62f3a19f676ccbadb9dbc5d07e951b2b26daf"},
     {file = "PyJWT-2.4.0.tar.gz", hash = "sha256:d42908208c699b3b973cbeb01a969ba6a96c821eefb1c5bfe4c390c01d67abba"},
@@ -2504,6 +2581,7 @@ version = "3.0.9"
 description = "pyparsing module - Classes and methods to define and execute parsing grammars"
 optional = false
 python-versions = ">=3.6.8"
+groups = ["main"]
 files = [
     {file = "pyparsing-3.0.9-py3-none-any.whl", hash = "sha256:5026bae9a10eeaefb61dab2f09052b9f4307d44aee4eda64b309723d8d206bbc"},
     {file = "pyparsing-3.0.9.tar.gz", hash = "sha256:2b020ecf7d21b687f219b71ecad3631f644a47f01403fa1d1036b0c6416d70fb"},
@@ -2518,6 +2596,7 @@ version = "0.18.1"
 description = "Persistent/Functional/Immutable data structures"
 optional = false
 python-versions = ">=3.7"
+groups = ["main"]
 files = [
     {file = "pyrsistent-0.18.1-cp310-cp310-macosx_10_9_universal2.whl", hash = "sha256:df46c854f490f81210870e509818b729db4488e1f30f2a1ce1698b2295a878d1"},
     {file = "pyrsistent-0.18.1-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:5d45866ececf4a5fff8742c25722da6d4c9e180daa7b405dc0a2a2790d668c26"},
@@ -2548,6 +2627,7 @@ version = "7.4.4"
 description = "pytest: simple powerful testing with Python"
 optional = false
 python-versions = ">=3.7"
+groups = ["main"]
 files = [
     {file = "pytest-7.4.4-py3-none-any.whl", hash = "sha256:b090cdf5ed60bf4c45261be03239c2c1c22df034fbffe691abe93cd80cea01d8"},
     {file = "pytest-7.4.4.tar.gz", hash = "sha256:2cf0005922c6ace4a3e2ec8b4080eb0d9753fdc93107415332f50ce9e7994280"},
@@ -2568,6 +2648,7 @@ version = "0.21.0"
 description = "Pytest support for asyncio"
 optional = false
 python-versions = ">=3.7"
+groups = ["main"]
 files = [
     {file = "pytest-asyncio-0.21.0.tar.gz", hash = "sha256:2b38a496aef56f56b0e87557ec313e11e1ab9276fc3863f6a7be0f1d0e415e1b"},
     {file = "pytest_asyncio-0.21.0-py3-none-any.whl", hash = "sha256:f2b3366b7cd501a4056858bd39349d5af19742aed2d81660b7998b6341c7eb9c"},
@@ -2586,6 +2667,7 @@ version = "1.0.8"
 description = "pytest-httpserver is a httpserver for pytest"
 optional = false
 python-versions = ">=3.8,<4.0"
+groups = ["main"]
 files = [
     {file = "pytest_httpserver-1.0.8-py3-none-any.whl", hash = "sha256:24cd3d9f6a0b927c7bfc400d0b3fda7442721b8267ce29942bf307b190f0bb09"},
     {file = "pytest_httpserver-1.0.8.tar.gz", hash = "sha256:e052f69bc8a9073db02484681e8e47004dd1fb3763b0ae833bd899e5895c559a"},
@@ -2600,6 +2682,7 @@ version = "0.6.3"
 description = "It helps to use fixtures in pytest.mark.parametrize"
 optional = false
 python-versions = "*"
+groups = ["main"]
 files = [
     {file = "pytest-lazy-fixture-0.6.3.tar.gz", hash = "sha256:0e7d0c7f74ba33e6e80905e9bfd81f9d15ef9a790de97993e34213deb5ad10ac"},
     {file = "pytest_lazy_fixture-0.6.3-py3-none-any.whl", hash = "sha256:e0b379f38299ff27a653f03eaa69b08a6fd4484e46fd1c9907d984b9f9daeda6"},
@@ -2614,6 +2697,7 @@ version = "1.1.0"
 description = "pytest plugin to run your tests in a specific order"
 optional = false
 python-versions = ">=3.6"
+groups = ["main"]
 files = [
     {file = "pytest-order-1.1.0.tar.gz", hash = "sha256:139d25b30826b78eebb42722f747eab14c44b88059d7a71d4f79d14a057269a5"},
     {file = "pytest_order-1.1.0-py3-none-any.whl", hash = "sha256:3b3730969c97900fa5cd31ecff80847680ed56b2490954565c14949ba60d9371"},
@@ -2628,6 +2712,7 @@ version = "0.9.3"
 description = "pytest plugin for repeating tests"
 optional = false
 python-versions = ">=3.7"
+groups = ["main"]
 files = [
     {file = "pytest_repeat-0.9.3-py3-none-any.whl", hash = "sha256:26ab2df18226af9d5ce441c858f273121e92ff55f5bb311d25755b8d7abdd8ed"},
     {file = "pytest_repeat-0.9.3.tar.gz", hash = "sha256:ffd3836dfcd67bb270bec648b330e20be37d2966448c4148c4092d1e8aba8185"},
@@ -2642,6 +2727,7 @@ version = "15.0"
 description = "pytest plugin to re-run tests to eliminate flaky failures"
 optional = false
 python-versions = ">=3.9"
+groups = ["main"]
 files = [
     {file = "pytest-rerunfailures-15.0.tar.gz", hash = "sha256:2d9ac7baf59f4c13ac730b47f6fa80e755d1ba0581da45ce30b72fb3542b4474"},
     {file = "pytest_rerunfailures-15.0-py3-none-any.whl", hash = "sha256:dd150c4795c229ef44320adc9a0c0532c51b78bb7a6843a8c53556b9a611df1a"},
@@ -2657,6 +2743,7 @@ version = "0.8.1"
 description = "Pytest plugin which splits the test suite to equally sized sub suites based on test execution time."
 optional = false
 python-versions = ">=3.7.1,<4.0"
+groups = ["main"]
 files = [
     {file = "pytest_split-0.8.1-py3-none-any.whl", hash = "sha256:74b110ea091bd147cc1c5f9665a59506e5cedfa66f96a89fb03e4ab447c2c168"},
     {file = "pytest_split-0.8.1.tar.gz", hash = "sha256:2d88bd3dc528689a7a3f58fc12ea165c3aa62e90795e420dfad920afe5612d6d"},
@@ -2671,6 +2758,7 @@ version = "2.1.0"
 description = "pytest plugin to abort hanging tests"
 optional = false
 python-versions = ">=3.6"
+groups = ["main"]
 files = [
     {file = "pytest-timeout-2.1.0.tar.gz", hash = "sha256:c07ca07404c612f8abbe22294b23c368e2e5104b521c1790195561f37e1ac3d9"},
     {file = "pytest_timeout-2.1.0-py3-none-any.whl", hash = "sha256:f6f50101443ce70ad325ceb4473c4255e9d74e3c7cd0ef827309dfa4c0d975c6"},
@@ -2685,6 +2773,7 @@ version = "3.3.1"
 description = "pytest xdist plugin for distributed testing, most importantly across multiple CPUs"
 optional = false
 python-versions = ">=3.7"
+groups = ["main"]
 files = [
     {file = "pytest-xdist-3.3.1.tar.gz", hash = "sha256:d5ee0520eb1b7bcca50a60a518ab7a7707992812c578198f8b44fdfac78e8c93"},
     {file = "pytest_xdist-3.3.1-py3-none-any.whl", hash = "sha256:ff9daa7793569e6a68544850fd3927cd257cc03a7ef76c95e86915355e82b5f2"},
@@ -2705,6 +2794,7 @@ version = "2.8.2"
 description = "Extensions to the standard Python datetime module"
 optional = false
 python-versions = "!=3.0.*,!=3.1.*,!=3.2.*,>=2.7"
+groups = ["main"]
 files = [
     {file = "python-dateutil-2.8.2.tar.gz", hash = "sha256:0123cacc1627ae19ddf3c27a5de5bd67ee4586fbdd6440d9748f8abb483d3e86"},
     {file = "python_dateutil-2.8.2-py2.py3-none-any.whl", hash = "sha256:961d03dc3453ebbc59dbdea9e4e11c5651520a876d0f4db161e8674aae935da9"},
@@ -2719,6 +2809,7 @@ version = "1.0.1"
 description = "Read key-value pairs from a .env file and set them as environment variables"
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "python-dotenv-1.0.1.tar.gz", hash = "sha256:e324ee90a023d808f1959c46bcbc04446a10ced277783dc6ee09987c37ec10ca"},
     {file = "python_dotenv-1.0.1-py3-none-any.whl", hash = "sha256:f7b63ef50f1b690dddf550d03497b66d609393b40b564ed0d674909a68ebf16a"},
@@ -2733,6 +2824,7 @@ version = "2024.1"
 description = "World timezone definitions, modern and historical"
 optional = false
 python-versions = "*"
+groups = ["main"]
 files = [
     {file = "pytz-2024.1-py2.py3-none-any.whl", hash = "sha256:328171f4e3623139da4983451950b28e95ac706e13f3f2630a879749e7a8b319"},
     {file = "pytz-2024.1.tar.gz", hash = "sha256:2a29735ea9c18baf14b448846bde5a48030ed267578472d8955cd0e7443a9812"},
@@ -2744,6 +2836,8 @@ version = "308"
 description = "Python for Window Extensions"
 optional = false
 python-versions = "*"
+groups = ["main"]
+markers = "sys_platform == \"win32\""
 files = [
     {file = "pywin32-308-cp310-cp310-win32.whl", hash = "sha256:796ff4426437896550d2981b9c2ac0ffd75238ad9ea2d3bfa67a1abd546d262e"},
     {file = "pywin32-308-cp310-cp310-win_amd64.whl", hash = "sha256:4fc888c59b3c0bef905ce7eb7e2106a07712015ea1c8234b703a088d46110e8e"},
@@ -2771,6 +2865,7 @@ version = "6.0.2"
 description = "YAML parser and emitter for Python"
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "PyYAML-6.0.2-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:0a9a2848a5b7feac301353437eb7d5957887edbf81d56e903999a75a3d743086"},
     {file = "PyYAML-6.0.2-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:29717114e51c84ddfba879543fb232a6ed60086602313ca38cce623c1d62cfbf"},
@@ -2833,6 +2928,7 @@ version = "2024.4.28"
 description = "Alternative regular expression module, to replace re."
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "regex-2024.4.28-cp310-cp310-macosx_10_9_universal2.whl", hash = "sha256:cd196d056b40af073d95a2879678585f0b74ad35190fac04ca67954c582c6b61"},
     {file = "regex-2024.4.28-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:8bb381f777351bd534462f63e1c6afb10a7caa9fa2a421ae22c26e796fe31b1f"},
@@ -2921,6 +3017,7 @@ version = "2.32.3"
 description = "Python HTTP for Humans."
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "requests-2.32.3-py3-none-any.whl", hash = "sha256:70761cfe03c773ceb22aa2f671b4757976145175cdfca038c02654d061d6dcc6"},
     {file = "requests-2.32.3.tar.gz", hash = "sha256:55365417734eb18255590a9ff9eb97e9e1da868d4ccd6402399eaf68af20a760"},
@@ -2942,6 +3039,7 @@ version = "0.25.3"
 description = "A utility library for mocking out the `requests` Python library."
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "responses-0.25.3-py3-none-any.whl", hash = "sha256:521efcbc82081ab8daa588e08f7e8a64ce79b91c39f6e62199b19159bea7dbcb"},
     {file = "responses-0.25.3.tar.gz", hash = "sha256:617b9247abd9ae28313d57a75880422d55ec63c29d33d629697590a034358dba"},
@@ -2961,6 +3059,7 @@ version = "0.1.4"
 description = "A pure python RFC3339 validator"
 optional = false
 python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*"
+groups = ["main"]
 files = [
     {file = "rfc3339_validator-0.1.4-py2.py3-none-any.whl", hash = "sha256:24f6ec1eda14ef823da9e36ec7113124b39c04d50a4d3d3a3c2859577e7791fa"},
     {file = "rfc3339_validator-0.1.4.tar.gz", hash = "sha256:138a2abdf93304ad60530167e51d2dfb9549521a836871b88d7f4695d0022f6b"},
@@ -2975,6 +3074,7 @@ version = "0.7.0"
 description = "An extremely fast Python linter and code formatter, written in Rust."
 optional = false
 python-versions = ">=3.7"
+groups = ["dev"]
 files = [
     {file = "ruff-0.7.0-py3-none-linux_armv6l.whl", hash = "sha256:0cdf20c2b6ff98e37df47b2b0bd3a34aaa155f59a11182c1303cce79be715628"},
     {file = "ruff-0.7.0-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:496494d350c7fdeb36ca4ef1c9f21d80d182423718782222c29b3e72b3512737"},
@@ -3002,6 +3102,7 @@ version = "0.10.0"
 description = "An Amazon S3 Transfer Manager"
 optional = false
 python-versions = ">= 3.8"
+groups = ["main"]
 files = [
     {file = "s3transfer-0.10.0-py3-none-any.whl", hash = "sha256:3cdb40f5cfa6966e812209d0994f2a4709b561c88e90cf00c2696d2df4e56b2e"},
     {file = "s3transfer-0.10.0.tar.gz", hash = "sha256:d0c8bbf672d5eebbe4e57945e23b972d963f07d82f661cabf678a5c88831595b"},
@@ -3019,6 +3120,7 @@ version = "1.0.4"
 description = "Classes implementing the SARIF 2.1.0 object model."
 optional = false
 python-versions = ">= 2.7"
+groups = ["main"]
 files = [
     {file = "sarif_om-1.0.4-py3-none-any.whl", hash = "sha256:539ef47a662329b1c8502388ad92457425e95dc0aaaf995fe46f4984c4771911"},
     {file = "sarif_om-1.0.4.tar.gz", hash = "sha256:cd5f416b3083e00d402a92e449a7ff67af46f11241073eea0461802a3b5aef98"},
@@ -3034,6 +3136,7 @@ version = "70.0.0"
 description = "Easily download, build, install, upgrade, and uninstall Python packages"
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "setuptools-70.0.0-py3-none-any.whl", hash = "sha256:54faa7f2e8d2d11bcd2c07bed282eef1046b5c080d1c32add737d7b5817b1ad4"},
     {file = "setuptools-70.0.0.tar.gz", hash = "sha256:f211a66637b8fa059bb28183da127d4e86396c991a942b028c6650d4319c3fd0"},
@@ -3049,6 +3152,7 @@ version = "1.16.0"
 description = "Python 2 and 3 compatibility utilities"
 optional = false
 python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*"
+groups = ["main"]
 files = [
     {file = "six-1.16.0-py2.py3-none-any.whl", hash = "sha256:8abb2f1d86890a2dfb989f9a77cfcfd3e47c2a354b01111771326f8aa26e0254"},
     {file = "six-1.16.0.tar.gz", hash = "sha256:1e61c37477a1626458e36f7b1d82aa5c9b094fa4802892072e49de9c60c4c926"},
@@ -3060,6 +3164,7 @@ version = "1.3.0"
 description = "Sniff out which async library your code is running under"
 optional = false
 python-versions = ">=3.7"
+groups = ["main"]
 files = [
     {file = "sniffio-1.3.0-py3-none-any.whl", hash = "sha256:eecefdce1e5bbfb7ad2eeaabf7c1eeb404d7757c379bd1f7e5cce9d8bf425384"},
     {file = "sniffio-1.3.0.tar.gz", hash = "sha256:e60305c5e5d314f5389259b7f22aaa33d8f7dee49763119234af3755c55b9101"},
@@ -3071,6 +3176,7 @@ version = "1.12"
 description = "Computer algebra system (CAS) in Python"
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "sympy-1.12-py3-none-any.whl", hash = "sha256:c3588cd4295d0c0f603d0f2ae780587e64e2efeedb3521e46b9bb1d08d184fa5"},
     {file = "sympy-1.12.tar.gz", hash = "sha256:ebf595c8dac3e0fdc4152c51878b498396ec7f30e7a914d6071e674d49420fb8"},
@@ -3085,6 +3191,7 @@ version = "4.9.0"
 description = "Python library for throwaway instances of anything that can run in a Docker container"
 optional = false
 python-versions = "<4.0,>=3.9"
+groups = ["main"]
 files = [
     {file = "testcontainers-4.9.0-py3-none-any.whl", hash = "sha256:c6fee929990972c40bf6b91b7072c94064ff3649b405a14fde0274c8b2479d32"},
     {file = "testcontainers-4.9.0.tar.gz", hash = "sha256:2cd6af070109ff68c1ab5389dc89c86c2dc3ab30a21ca734b2cb8f0f80ad479e"},
@@ -3138,6 +3245,7 @@ version = "0.10.2"
 description = "Python Library for Tom's Obvious, Minimal Language"
 optional = false
 python-versions = ">=2.6, !=3.0.*, !=3.1.*, !=3.2.*"
+groups = ["main"]
 files = [
     {file = "toml-0.10.2-py2.py3-none-any.whl", hash = "sha256:806143ae5bfb6a3c6e736a764057db0e6a0e05e338b5630894a5f779cabb4f9b"},
     {file = "toml-0.10.2.tar.gz", hash = "sha256:b3bda1d108d5dd99f4a20d24d9c348e91c4db7ab1b749200bded2f839ccbe68f"},
@@ -3149,6 +3257,7 @@ version = "1.5.0.20240925"
 description = "Typing stubs for jwcrypto"
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "types-jwcrypto-1.5.0.20240925.tar.gz", hash = "sha256:50e17b790378c96239344476c7bd13b52d0c7eeb6d16c2d53723e48cc6bbf4fe"},
     {file = "types_jwcrypto-1.5.0.20240925-py3-none-any.whl", hash = "sha256:2d12a2d528240d326075e896aafec7056b9136bf3207fa6ccf3fcb8fbf9e11a1"},
@@ -3163,6 +3272,7 @@ version = "5.9.5.12"
 description = "Typing stubs for psutil"
 optional = false
 python-versions = "*"
+groups = ["main"]
 files = [
     {file = "types-psutil-5.9.5.12.tar.gz", hash = "sha256:61a91679d3fe737250013b624dca09375e7cc3ad77dcc734553746c429c02aca"},
     {file = "types_psutil-5.9.5.12-py3-none-any.whl", hash = "sha256:e9a147b8561235c6afcce5aa1adb973fad9ab2c50cf89820697687f53510358f"},
@@ -3174,6 +3284,7 @@ version = "2.9.21.20241019"
 description = "Typing stubs for psycopg2"
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "types-psycopg2-2.9.21.20241019.tar.gz", hash = "sha256:bca89b988d2ebd19bcd08b177d22a877ea8b841decb10ed130afcf39404612fa"},
     {file = "types_psycopg2-2.9.21.20241019-py3-none-any.whl", hash = "sha256:44d091e67732d16a941baae48cd7b53bf91911bc36888652447cf1ef0c1fb3f6"},
@@ -3185,6 +3296,7 @@ version = "0.6.3.3"
 description = "Typing stubs for pytest-lazy-fixture"
 optional = false
 python-versions = "*"
+groups = ["main"]
 files = [
     {file = "types-pytest-lazy-fixture-0.6.3.3.tar.gz", hash = "sha256:2ef79d66bcde0e50acdac8dc55074b9ae0d4cfaeabdd638f5522f4cac7c8a2c7"},
     {file = "types_pytest_lazy_fixture-0.6.3.3-py3-none-any.whl", hash = "sha256:a56a55649147ff960ff79d4b2c781a4f769351abc1876873f3116d0bd0c96353"},
@@ -3196,6 +3308,7 @@ version = "6.0.12.20240917"
 description = "Typing stubs for PyYAML"
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "types-PyYAML-6.0.12.20240917.tar.gz", hash = "sha256:d1405a86f9576682234ef83bcb4e6fff7c9305c8b1fbad5e0bcd4f7dbdc9c587"},
     {file = "types_PyYAML-6.0.12.20240917-py3-none-any.whl", hash = "sha256:392b267f1c0fe6022952462bf5d6523f31e37f6cea49b14cee7ad634b6301570"},
@@ -3207,6 +3320,7 @@ version = "2.31.0.0"
 description = "Typing stubs for requests"
 optional = false
 python-versions = "*"
+groups = ["main"]
 files = [
     {file = "types-requests-2.31.0.0.tar.gz", hash = "sha256:c1c29d20ab8d84dff468d7febfe8e0cb0b4664543221b386605e14672b44ea25"},
     {file = "types_requests-2.31.0.0-py3-none-any.whl", hash = "sha256:7c5cea7940f8e92ec560bbc468f65bf684aa3dcf0554a6f8c4710f5f708dc598"},
@@ -3221,6 +3335,7 @@ version = "0.6.0.post3"
 description = "Type annotations and code completion for s3transfer"
 optional = false
 python-versions = ">=3.7,<4.0"
+groups = ["main"]
 files = [
     {file = "types-s3transfer-0.6.0.post3.tar.gz", hash = "sha256:92c3704e5d041202bfb5ddb79d083fd1a02de2c5dfec6a91576823e6b5c93993"},
     {file = "types_s3transfer-0.6.0.post3-py3-none-any.whl", hash = "sha256:eedc5117275565b3c83662c0ccc81662a34da5dda8bd502b89d296b6d5cb091d"},
@@ -3232,6 +3347,7 @@ version = "0.10.8.6"
 description = "Typing stubs for toml"
 optional = false
 python-versions = "*"
+groups = ["main"]
 files = [
     {file = "types-toml-0.10.8.6.tar.gz", hash = "sha256:6d3ac79e36c9ee593c5d4fb33a50cca0e3adceb6ef5cff8b8e5aef67b4c4aaf2"},
     {file = "types_toml-0.10.8.6-py3-none-any.whl", hash = "sha256:de7b2bb1831d6f7a4b554671ffe5875e729753496961b3e9b202745e4955dafa"},
@@ -3243,6 +3359,7 @@ version = "1.26.17"
 description = "Typing stubs for urllib3"
 optional = false
 python-versions = "*"
+groups = ["main"]
 files = [
     {file = "types-urllib3-1.26.17.tar.gz", hash = "sha256:73fd274524c3fc7cd8cd9ceb0cb67ed99b45f9cb2831013e46d50c1451044800"},
     {file = "types_urllib3-1.26.17-py3-none-any.whl", hash = "sha256:0d027fcd27dbb3cb532453b4d977e05bc1e13aefd70519866af211b3003d895d"},
@@ -3254,6 +3371,7 @@ version = "4.12.2"
 description = "Backported and Experimental Type Hints for Python 3.8+"
 optional = false
 python-versions = ">=3.8"
+groups = ["main", "dev"]
 files = [
     {file = "typing_extensions-4.12.2-py3-none-any.whl", hash = "sha256:04e5ca0351e0f3f85c6853954072df659d0d13fac324d0072316b67d7794700d"},
     {file = "typing_extensions-4.12.2.tar.gz", hash = "sha256:1a7ead55c7e559dd4dee8856e3a88b41225abfe1ce8df57b7c13915fe121ffb8"},
@@ -3265,6 +3383,7 @@ version = "1.26.19"
 description = "HTTP library with thread-safe connection pooling, file post, and more."
 optional = false
 python-versions = "!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*,!=3.4.*,!=3.5.*,>=2.7"
+groups = ["main"]
 files = [
     {file = "urllib3-1.26.19-py2.py3-none-any.whl", hash = "sha256:37a0344459b199fce0e80b0d3569837ec6b6937435c5244e7fd73fa6006830f3"},
     {file = "urllib3-1.26.19.tar.gz", hash = "sha256:3e3d753a8618b86d7de333b4223005f68720bcd6a7d2bcb9fbd2229ec7c1e429"},
@@ -3281,6 +3400,7 @@ version = "12.0"
 description = "An implementation of the WebSocket Protocol (RFC 6455 & 7692)"
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "websockets-12.0-cp310-cp310-macosx_10_9_universal2.whl", hash = "sha256:d554236b2a2006e0ce16315c16eaa0d628dab009c33b63ea03f41c6107958374"},
     {file = "websockets-12.0-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:2d225bb6886591b1746b17c0573e29804619c8f755b5598d875bb4235ea639be"},
@@ -3362,6 +3482,7 @@ version = "3.0.6"
 description = "The comprehensive WSGI web application library."
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "werkzeug-3.0.6-py3-none-any.whl", hash = "sha256:1bc0c2310d2fbb07b1dd1105eba2f7af72f322e1e455f2f93c993bee8c8a5f17"},
     {file = "werkzeug-3.0.6.tar.gz", hash = "sha256:a8dd59d4de28ca70471a34cba79bed5f7ef2e036a76b3ab0835474246eb41f8d"},
@@ -3379,6 +3500,7 @@ version = "1.14.1"
 description = "Module for decorators, wrappers and monkey patching."
 optional = false
 python-versions = "!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*,!=3.4.*,>=2.7"
+groups = ["main"]
 files = [
     {file = "wrapt-1.14.1-cp27-cp27m-macosx_10_9_x86_64.whl", hash = "sha256:1b376b3f4896e7930f1f772ac4b064ac12598d1c38d04907e696cc4d794b43d3"},
     {file = "wrapt-1.14.1-cp27-cp27m-manylinux1_i686.whl", hash = "sha256:903500616422a40a98a5a3c4ff4ed9d0066f3b4c951fa286018ecdf0750194ef"},
@@ -3462,6 +3584,7 @@ version = "0.13.0"
 description = "Makes working with XML feel like you are working with JSON"
 optional = false
 python-versions = ">=3.4"
+groups = ["main"]
 files = [
     {file = "xmltodict-0.13.0-py2.py3-none-any.whl", hash = "sha256:aa89e8fd76320154a40d19a0df04a4695fb9dc5ba977cbb68ab3e4eb225e7852"},
     {file = "xmltodict-0.13.0.tar.gz", hash = "sha256:341595a488e3e01a85a9d8911d8912fd922ede5fecc4dce437eb4b6c8d037e56"},
@@ -3473,6 +3596,7 @@ version = "1.17.2"
 description = "Yet another URL library"
 optional = false
 python-versions = ">=3.9"
+groups = ["main"]
 files = [
     {file = "yarl-1.17.2-cp310-cp310-macosx_10_9_universal2.whl", hash = "sha256:93771146ef048b34201bfa382c2bf74c524980870bb278e6df515efaf93699ff"},
     {file = "yarl-1.17.2-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:8281db240a1616af2f9c5f71d355057e73a1409c4648c8949901396dc0a3c151"},
@@ -3569,6 +3693,7 @@ version = "0.23.0"
 description = "Zstandard bindings for Python"
 optional = false
 python-versions = ">=3.8"
+groups = ["main"]
 files = [
     {file = "zstandard-0.23.0-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:bf0a05b6059c0528477fba9054d09179beb63744355cab9f38059548fedd46a9"},
     {file = "zstandard-0.23.0-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:fc9ca1c9718cb3b06634c7c8dec57d24e9438b2aa9a0f02b8bb36bf478538880"},
@@ -3676,6 +3801,6 @@ cffi = {version = ">=1.11", markers = "platform_python_implementation == \"PyPy\
 cffi = ["cffi (>=1.11)"]
 
 [metadata]
-lock-version = "2.0"
+lock-version = "2.1"
 python-versions = "^3.11"
-content-hash = "b25702660d965fa9f21db0b20fa07b545d199d8ccb303c67eece77b338b08494"
+content-hash = "4dc3165fe22c0e0f7a030ea0f8a680ae2ff74561d8658c393abbe9112caaf5d7"

pyproject.toml

@@ -17,7 +17,7 @@ Jinja2 = "^3.1.5"
 types-requests = "^2.31.0.0"
 types-psycopg2 = "^2.9.21.20241019"
 boto3 = "^1.34.11"
-boto3-stubs = {extras = ["s3", "kms"], version = "^1.26.16"}
+boto3-stubs = {extras = ["s3"], version = "^1.26.16"}
 moto = {extras = ["server"], version = "^5.0.6"}
 backoff = "^2.2.1"
 pytest-lazy-fixture = "^0.6.3"
@@ -43,7 +43,7 @@ websockets = "^12.0"
 clickhouse-connect = "^0.7.16"
 kafka-python = "^2.0.2"
 jwcrypto = "^1.5.6"
-h2 = "^4.1.0"
+h2 = {git = "https://github.com/python-hyper/h2"}
 types-jwcrypto = "^1.5.0.20240925"
 pyyaml = "^6.0.2"
 types-pyyaml = "^6.0.12.20240917"
@@ -94,6 +94,7 @@ target-version = "py311"
 extend-exclude = [
     "vendor/",
     "target/",
+    "test_runner/stubs/", # Autogenerated by mypy's stubgen
 ]
 line-length = 100 # this setting is rather guidance, it won't fail if it can't make the shorter
 

safekeeper/client/src/mgmt_api.rs

@@ -4,7 +4,7 @@
 //! united.
 
 use reqwest::{IntoUrl, Method, StatusCode};
-use safekeeper_api::models::TimelineStatus;
+use safekeeper_api::models::{TimelineCreateRequest, TimelineStatus};
 use std::error::Error as _;
 use utils::{
     http::error::HttpErrorBody,
@@ -76,6 +76,28 @@ impl Client {
         }
     }
 
+    pub async fn create_timeline(&self, req: &TimelineCreateRequest) -> Result<TimelineStatus> {
+        let uri = format!(
+            "{}/v1/tenant/{}/timeline/{}",
+            self.mgmt_api_endpoint, req.tenant_id, req.timeline_id
+        );
+        let resp = self.post(&uri, req).await?;
+        resp.json().await.map_err(Error::ReceiveBody)
+    }
+
+    pub async fn delete_timeline(
+        &self,
+        tenant_id: TenantId,
+        timeline_id: TimelineId,
+    ) -> Result<TimelineStatus> {
+        let uri = format!(
+            "{}/v1/tenant/{}/timeline/{}",
+            self.mgmt_api_endpoint, tenant_id, timeline_id
+        );
+        let resp = self.request(Method::DELETE, &uri, ()).await?;
+        resp.json().await.map_err(Error::ReceiveBody)
+    }
+
     pub async fn timeline_status(
         &self,
         tenant_id: TenantId,
@@ -107,6 +129,14 @@ impl Client {
         self.get(&uri).await
     }
 
+    async fn post<B: serde::Serialize, U: IntoUrl>(
+        &self,
+        uri: U,
+        body: B,
+    ) -> Result<reqwest::Response> {
+        self.request(Method::POST, uri, body).await
+    }
+
     async fn get<U: IntoUrl>(&self, uri: U) -> Result<reqwest::Response> {
         self.request(Method::GET, uri, ()).await
     }

scripts/comment-test-report.js

@@ -84,6 +84,12 @@ const parseReportJson = async ({ reportJsonUrl, fetch }) => {
                 } else {
                     arch = "unknown"
                 }
+                let lfcState = ""
+                if (test.parameters.includes("'with-lfc'")) {
+                    lfcState = "with-lfc"
+                } else {
+                    lfcState = "without-lfc"
+                }
 
                 // Removing build type and PostgreSQL version from the test name to make it shorter
                 const testName = test.name.replace(new RegExp(`${buildType}-pg${pgVersion}-?`), "").replace("[]", "")
@@ -91,6 +97,7 @@ const parseReportJson = async ({ reportJsonUrl, fetch }) => {
                 test.pgVersion = pgVersion
                 test.buildType = buildType
                 test.arch = arch
+                test.lfcState = lfcState
 
                 if (test.status === "passed") {
                     passedTests[pgVersion][testName].push(test)
@@ -157,7 +164,7 @@ const reportSummary = async (params) => {
                 const links = []
                 for (const test of tests) {
                     const allureLink = `${reportUrl}#suites/${test.parentUid}/${test.uid}`
-                    links.push(`[${test.buildType}-${test.arch}](${allureLink})`)
+                    links.push(`[${test.buildType}-${test.arch}-${test.lfcState}](${allureLink})`)
                 }
                 summary += `- \`${testName}\`: ${links.join(", ")}\n`
             }
@@ -188,7 +195,7 @@ const reportSummary = async (params) => {
                     const links = []
                     for (const test of tests) {
                         const allureLink = `${reportUrl}#suites/${test.parentUid}/${test.uid}/retries`
-                        links.push(`[${test.buildType}-${test.arch}](${allureLink})`)
+                        links.push(`[${test.buildType}-${test.arch}-${test.lfcState}](${allureLink})`)
                     }
                     summary += `- \`${testName}\`: ${links.join(", ")}\n`
                 }

scripts/ingest_regress_test_result-new-format.py

@@ -134,7 +134,7 @@ def ingest_test_result(
             if p["name"].startswith("__")
         }
         arch = parameters.get("arch", "UNKNOWN").strip("'")
-        lfc = parameters.get("lfc", "False") == "True"
+        lfc = parameters.get("lfc", "without-lfc").strip("'") == "with-lfc"
 
         build_type, pg_version, unparametrized_name = parse_test_name(test["name"])
         labels = {label["name"]: label["value"] for label in test["labels"]}

storage_controller/Cargo.toml

@@ -45,12 +45,11 @@ strum_macros.workspace = true
 
 diesel = { version = "2.2.6", features = [
     "serde_json",
-    "postgres",
-    "r2d2",
     "chrono",
 ] }
+diesel-async = { version = "0.5.2", features = ["postgres", "bb8", "async-connection-wrapper"] }
 diesel_migrations = { version = "2.2.0" }
-r2d2 = { version = "0.8.10" }
+scoped-futures = "0.1.4"
 
 utils = { path = "../libs/utils/" }
 metrics = { path = "../libs/metrics/" }

storage_controller/src/main.rs

@@ -308,7 +308,7 @@ async fn async_main() -> anyhow::Result<()> {
     // Validate that we can connect to the database
     Persistence::await_connection(&secrets.database_url, args.db_connect_timeout.into()).await?;
 
-    let persistence = Arc::new(Persistence::new(secrets.database_url));
+    let persistence = Arc::new(Persistence::new(secrets.database_url).await);
 
     let service = Service::spawn(config, persistence.clone()).await?;
 

test_runner/fixtures/fast_import.py

@@ -50,9 +50,8 @@ class FastImport(AbstractNeonCli):
 
     def run(
         self,
-        pg_port: int | None = None,
+        pg_port: int,
         source_connection_string: str | None = None,
-        restore_connection_string: str | None = None,
         s3prefix: str | None = None,
         interactive: bool = False,
     ) -> subprocess.CompletedProcess[str]:
@@ -61,14 +60,11 @@ class FastImport(AbstractNeonCli):
         args = [
             f"--pg-bin-dir={self.pg_bin}",
             f"--pg-lib-dir={self.pg_lib}",
+            f"--pg-port={pg_port}",
             f"--working-directory={self.workdir}",
         ]
-        if pg_port is not None:
-            args.append(f"--pg-port={pg_port}")
         if source_connection_string is not None:
             args.append(f"--source-connection-string={source_connection_string}")
-        if restore_connection_string is not None:
-            args.append(f"--restore-connection-string={restore_connection_string}")
         if s3prefix is not None:
             args.append(f"--s3-prefix={s3prefix}")
         if interactive:

test_runner/fixtures/metrics.py

@@ -165,7 +165,6 @@ PAGESERVER_PER_TENANT_METRICS: tuple[str, ...] = (
     "pageserver_evictions_with_low_residence_duration_total",
     "pageserver_aux_file_estimated_size",
     "pageserver_valid_lsn_lease_count",
-    "pageserver_flush_wait_upload_seconds",
     counter("pageserver_tenant_throttling_count_accounted_start"),
     counter("pageserver_tenant_throttling_count_accounted_finish"),
     counter("pageserver_tenant_throttling_wait_usecs_sum"),

test_runner/fixtures/neon_cli.py

@@ -523,6 +523,7 @@ class NeonLocalCli(AbstractNeonCli):
         remote_ext_config: str | None = None,
         pageserver_id: int | None = None,
         allow_multiple: bool = False,
+        create_test_user: bool = False,
         basebackup_request_tries: int | None = None,
         env: dict[str, str] | None = None,
     ) -> subprocess.CompletedProcess[str]:
@@ -544,6 +545,8 @@ class NeonLocalCli(AbstractNeonCli):
             args.extend(["--pageserver-id", str(pageserver_id)])
         if allow_multiple:
             args.extend(["--allow-multiple"])
+        if create_test_user:
+            args.extend(["--create-test-user"])
 
         res = self.raw_cli(args, extra_env_vars)
         res.check_returncode()

test_runner/fixtures/neon_fixtures.py

@@ -26,7 +26,6 @@ from urllib.parse import quote, urlparse
 
 import asyncpg
 import backoff
-import boto3
 import httpx
 import psycopg2
 import psycopg2.sql
@@ -37,8 +36,6 @@ from _pytest.config import Config
 from _pytest.config.argparsing import Parser
 from _pytest.fixtures import FixtureRequest
 from jwcrypto import jwk
-from mypy_boto3_kms import KMSClient
-from mypy_boto3_s3 import S3Client
 
 # Type-related stuff
 from psycopg2.extensions import connection as PgConnection
@@ -201,30 +198,6 @@ def mock_s3_server(port_distributor: PortDistributor) -> Iterator[MockS3Server]:
     mock_s3_server.kill()
 
 
-@pytest.fixture(scope="session")
-def mock_kms(mock_s3_server: MockS3Server) -> Iterator[KMSClient]:
-    yield boto3.client(
-        "kms",
-        endpoint_url=mock_s3_server.endpoint(),
-        region_name=mock_s3_server.region(),
-        aws_access_key_id=mock_s3_server.access_key(),
-        aws_secret_access_key=mock_s3_server.secret_key(),
-        aws_session_token=mock_s3_server.session_token(),
-    )
-
-
-@pytest.fixture(scope="session")
-def mock_s3_client(mock_s3_server: MockS3Server) -> Iterator[S3Client]:
-    yield boto3.client(
-        "s3",
-        endpoint_url=mock_s3_server.endpoint(),
-        region_name=mock_s3_server.region(),
-        aws_access_key_id=mock_s3_server.access_key(),
-        aws_secret_access_key=mock_s3_server.secret_key(),
-        aws_session_token=mock_s3_server.session_token(),
-    )
-
-
 class PgProtocol:
     """Reusable connection logic"""
 
@@ -3945,6 +3918,7 @@ class Endpoint(PgProtocol, LogUtils):
         pageserver_id: int | None = None,
         safekeepers: list[int] | None = None,
         allow_multiple: bool = False,
+        create_test_user: bool = False,
         basebackup_request_tries: int | None = None,
         env: dict[str, str] | None = None,
     ) -> Self:
@@ -3966,6 +3940,7 @@ class Endpoint(PgProtocol, LogUtils):
             remote_ext_config=remote_ext_config,
             pageserver_id=pageserver_id,
             allow_multiple=allow_multiple,
+            create_test_user=create_test_user,
             basebackup_request_tries=basebackup_request_tries,
             env=env,
         )
@@ -4415,6 +4390,7 @@ class Safekeeper(LogUtils):
                 "1s",
                 "--eviction-min-resident",
                 "10s",
+                "--wal-reader-fanout",
             ]
 
         self.extra_opts = extra_opts

test_runner/fixtures/pageserver/allowed_errors.py

@@ -99,8 +99,11 @@ DEFAULT_PAGESERVER_ALLOWED_ERRORS = (
     ".*WARN.*path=/v1/utilization .*request was dropped before completing",
     # Can happen during shutdown
     ".*scheduling deletion on drop failed: queue is in state Stopped.*",
-    # Too many frozen layers error is normal during intensive benchmarks
-    ".*too many frozen layers.*",
+    # L0 flush backpressure delays are expected under heavy ingest load. We want to exercise
+    # this backpressure in tests.
+    ".*delaying layer flush by \\S+ for compaction backpressure.*",
+    ".*stalling layer flushes for compaction backpressure.*",
+    ".*layer roll waiting for flush due to compaction backpressure.*",
 )
 
 

test_runner/fixtures/parametrize.py

@@ -121,6 +121,8 @@ def pytest_runtest_makereport(*args, **kwargs):
     }.get(os.uname().machine, "UNKNOWN")
     arch = os.getenv("RUNNER_ARCH", uname_m)
     allure.dynamic.parameter("__arch", arch)
-    allure.dynamic.parameter("__lfc", os.getenv("USE_LFC") != "false")
+    allure.dynamic.parameter(
+        "__lfc", "with-lfc" if os.getenv("USE_LFC") != "false" else "without-lfc"
+    )
 
     yield

test_runner/performance/test_layer_map.py

@@ -23,6 +23,8 @@ def test_layer_map(neon_env_builder: NeonEnvBuilder, zenbenchmark):
             "checkpoint_distance": "16384",
             "compaction_period": "1 s",
             "compaction_threshold": "1",
+            "l0_flush_delay_threshold": "0",
+            "l0_flush_stall_threshold": "0",
             "compaction_target_size": "16384",
         }
     )

test_runner/regress/test_attach_tenant_config.py

@@ -139,6 +139,8 @@ def test_fully_custom_config(positive_env: NeonEnv):
     fully_custom_config = {
         "compaction_period": "1h",
         "compaction_threshold": 13,
+        "l0_flush_delay_threshold": 25,
+        "l0_flush_stall_threshold": 42,
         "compaction_target_size": 1048576,
         "checkpoint_distance": 10000,
         "checkpoint_timeout": "13m",

test_runner/regress/test_branch_and_gc.py

@@ -64,6 +64,8 @@ def test_branch_and_gc(neon_simple_env: NeonEnv):
             # tweak the default settings to allow quickly create image layers and L1 layers
             "compaction_period": "1 s",
             "compaction_threshold": "2",
+            "l0_flush_delay_threshold": "20",
+            "l0_flush_stall_threshold": "40",
             "image_creation_threshold": "1",
             # Disable PITR, this test will set an explicit space-based GC limit
             "pitr_interval": "0 s",

test_runner/regress/test_branching.py

@@ -19,6 +19,7 @@ from fixtures.pageserver.utils import wait_until_tenant_active
 from fixtures.utils import query_scalar
 from performance.test_perf_pgbench import get_scales_matrix
 from requests import RequestException
+from requests.exceptions import RetryError
 
 
 # Test branch creation
@@ -176,11 +177,8 @@ def test_cannot_create_endpoint_on_non_uploaded_timeline(neon_env_builder: NeonE
 
         env.neon_cli.mappings_map_branch(initial_branch, env.initial_tenant, env.initial_timeline)
 
-        with pytest.raises(RuntimeError, match="ERROR: Not found: Timeline"):
-            env.endpoints.create_start(
-                initial_branch, tenant_id=env.initial_tenant, basebackup_request_tries=2
-            )
-        ps_http.configure_failpoints(("before-upload-index-pausable", "off"))
+        with pytest.raises(RuntimeError, match="is not active, state: Loading"):
+            env.endpoints.create_start(initial_branch, tenant_id=env.initial_tenant)
     finally:
         env.pageserver.stop(immediate=True)
 
@@ -221,10 +219,7 @@ def test_cannot_branch_from_non_uploaded_branch(neon_env_builder: NeonEnvBuilder
 
         branch_id = TimelineId.generate()
 
-        with pytest.raises(
-            PageserverApiException,
-            match="Cannot branch off the timeline that's not present in pageserver",
-        ):
+        with pytest.raises(RetryError, match="too many 503 error responses"):
             ps_http.timeline_create(
                 env.pg_version,
                 env.initial_tenant,

test_runner/regress/test_compaction.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import json
+import math
+import random
 import time
 from enum import StrEnum
 
@@ -128,11 +130,6 @@ def test_pageserver_gc_compaction_smoke(neon_env_builder: NeonEnvBuilder, with_b
     }
 
     env = neon_env_builder.init_start(initial_tenant_conf=SMOKE_CONF)
-    env.pageserver.allowed_errors.append(
-        r".*failed to acquire partition lock during gc-compaction.*"
-    )
-    env.pageserver.allowed_errors.append(r".*repartition() called concurrently.*")
-
     tenant_id = env.initial_tenant
     timeline_id = env.initial_timeline
 
@@ -147,11 +144,14 @@ def test_pageserver_gc_compaction_smoke(neon_env_builder: NeonEnvBuilder, with_b
     log.info("Writing initial data ...")
     workload.write_rows(row_count, env.pageserver.id)
 
+    ps_http.timeline_gc(
+        tenant_id, timeline_id, None
+    )  # Force refresh gc info to have gc_cutoff generated
+
     child_workloads: list[Workload] = []
 
     for i in range(1, churn_rounds + 1):
-        if i % 10 == 0:
-            log.info(f"Running churn round {i}/{churn_rounds} ...")
+        log.info(f"Running churn round {i}/{churn_rounds} ...")
         if i % 10 == 5 and with_branches == "with_branches":
             branch_name = f"child-{i}"
             branch_timeline_id = env.create_branch(branch_name)
@@ -172,8 +172,10 @@ def test_pageserver_gc_compaction_smoke(neon_env_builder: NeonEnvBuilder, with_b
                     "sub_compaction_max_job_size_mb": 16,
                 },
             )
-
-        workload.churn_rows(row_count, env.pageserver.id)
+        # do not wait for upload so that we can see if gc_compaction works well with data being ingested
+        workload.churn_rows(row_count, env.pageserver.id, upload=False)
+        time.sleep(1)
+        workload.validate(env.pageserver.id)
 
     def compaction_finished():
         queue_depth = len(ps_http.timeline_compact_info(tenant_id, timeline_id))
@@ -197,6 +199,230 @@ def test_pageserver_gc_compaction_smoke(neon_env_builder: NeonEnvBuilder, with_b
     ps_http.timeline_gc(tenant_id, timeline_id, None)
 
 
+@pytest.mark.parametrize(
+    "compaction_mode",
+    ["before_restart", "after_restart"],
+)
+def test_pageserver_gc_compaction_idempotent(
+    neon_env_builder: NeonEnvBuilder, compaction_mode: str
+):
+    """
+    Do gc-compaction twice without writing any new data and see if anything breaks.
+    We run this test in two modes:
+    - before_restart: run two gc-compactions before pageserver restart
+    - after_restart: run one gc-compaction before and one after pageserver restart
+    """
+    SMOKE_CONF = {
+        # Run both gc and gc-compaction.
+        "gc_period": "5s",
+        "compaction_period": "5s",
+        # No PiTR interval and small GC horizon
+        "pitr_interval": "0s",
+        "gc_horizon": 1024,
+        "lsn_lease_length": "0s",
+    }
+
+    env = neon_env_builder.init_start(initial_tenant_conf=SMOKE_CONF)
+    tenant_id = env.initial_tenant
+    timeline_id = env.initial_timeline
+
+    # Only in testing mode: the warning is expected because we rewrite a layer file of different generations.
+    # We could potentially patch the sanity-check code to not emit the warning in the future.
+    env.pageserver.allowed_errors.append(".*was unlinked but was not dangling.*")
+
+    row_count = 10000
+
+    ps_http = env.pageserver.http_client()
+
+    workload = Workload(env, tenant_id, timeline_id)
+    workload.init(env.pageserver.id)
+
+    workload.write_rows(row_count, env.pageserver.id)
+
+    child_workloads: list[Workload] = []
+
+    def compaction_finished():
+        queue_depth = len(ps_http.timeline_compact_info(tenant_id, timeline_id))
+        assert queue_depth == 0
+
+    workload.churn_rows(row_count, env.pageserver.id)
+    env.create_branch("child_branch")  # so that we have a retain_lsn
+    workload.churn_rows(row_count, env.pageserver.id)
+    # compact 3 times if mode is before_restart
+    n_compactions = 3 if compaction_mode == "before_restart" else 1
+    for _ in range(n_compactions):
+        # Force refresh gc info to have gc_cutoff generated
+        ps_http.timeline_gc(tenant_id, timeline_id, None)
+        ps_http.timeline_compact(
+            tenant_id,
+            timeline_id,
+            enhanced_gc_bottom_most_compaction=True,
+            body={
+                "scheduled": True,
+                "sub_compaction": True,
+                "compact_key_range": {
+                    "start": "000000000000000000000000000000000000",
+                    "end": "030000000000000000000000000000000000",
+                },
+                "sub_compaction_max_job_size_mb": 16,
+            },
+        )
+        wait_until(compaction_finished, timeout=60)
+    if compaction_mode == "after_restart":
+        env.pageserver.restart(True)
+        ps_http.timeline_gc(
+            tenant_id, timeline_id, None
+        )  # Force refresh gc info to have gc_cutoff generated
+        for _ in range(3):
+            ps_http.timeline_compact(
+                tenant_id,
+                timeline_id,
+                enhanced_gc_bottom_most_compaction=True,
+                body={
+                    "scheduled": True,
+                    "sub_compaction": True,
+                    "compact_key_range": {
+                        "start": "000000000000000000000000000000000000",
+                        "end": "030000000000000000000000000000000000",
+                    },
+                    "sub_compaction_max_job_size_mb": 16,
+                },
+            )
+            wait_until(compaction_finished, timeout=60)
+
+    # ensure gc_compaction is scheduled and it's actually running (instead of skipping due to no layers picked)
+    env.pageserver.assert_log_contains(
+        "scheduled_compact_timeline.*picked .* layers for compaction"
+    )
+
+    # ensure we hit the duplicated layer key warning at least once: we did two compactions consecutively,
+    # and the second one should have hit the duplicated layer key warning.
+    if compaction_mode == "before_restart":
+        env.pageserver.assert_log_contains("duplicated layer key in the same generation")
+    else:
+        env.pageserver.assert_log_contains("same layer key at different generation")
+
+    log.info("Validating at workload end ...")
+    workload.validate(env.pageserver.id)
+    for child_workload in child_workloads:
+        log.info(f"Validating at branch {child_workload.branch_name}")
+        child_workload.validate(env.pageserver.id)
+
+    # Run a legacy compaction+gc to ensure gc-compaction can coexist with legacy compaction.
+    ps_http.timeline_checkpoint(tenant_id, timeline_id, wait_until_uploaded=True)
+    ps_http.timeline_gc(tenant_id, timeline_id, None)
+
+
+@skip_in_debug_build("only run with release build")
+def test_pageserver_gc_compaction_interrupt(neon_env_builder: NeonEnvBuilder):
+    """
+    Force interrupt a gc-compaction and see if anything breaks.
+    """
+    SMOKE_CONF = {
+        # Run both gc and gc-compaction.
+        "gc_period": "5s",
+        "compaction_period": "5s",
+        # No PiTR interval and small GC horizon
+        "pitr_interval": "0s",
+        "gc_horizon": "1024",
+        "lsn_lease_length": "0s",
+    }
+
+    env = neon_env_builder.init_start(initial_tenant_conf=SMOKE_CONF)
+    tenant_id = env.initial_tenant
+    timeline_id = env.initial_timeline
+
+    # Only in testing mode: the warning is expected because we rewrite a layer file of different generations.
+    # We could potentially patch the sanity-check code to not emit the warning in the future.
+    env.pageserver.allowed_errors.append(".*was unlinked but was not dangling.*")
+
+    row_count = 10000
+    churn_rounds = 20
+
+    ps_http = env.pageserver.http_client()
+
+    workload = Workload(env, tenant_id, timeline_id)
+    workload.init(env.pageserver.id)
+
+    log.info("Writing initial data ...")
+    workload.write_rows(row_count, env.pageserver.id)
+
+    def compaction_finished():
+        queue_depth = len(ps_http.timeline_compact_info(tenant_id, timeline_id))
+        assert queue_depth == 0
+
+    expected_compaction_time_seconds = 5.0
+    ps_http.timeline_gc(
+        tenant_id, timeline_id, None
+    )  # Force refresh gc info to have gc_cutoff generated
+    for i in range(1, churn_rounds + 1):
+        log.info(f"Running churn round {i}/{churn_rounds} ...")
+        workload.churn_rows(row_count, env.pageserver.id)
+        ps_http.timeline_compact(
+            tenant_id,
+            timeline_id,
+            enhanced_gc_bottom_most_compaction=True,
+            body={
+                "scheduled": True,
+                "sub_compaction": True,
+                "compact_key_range": {
+                    "start": "000000000000000000000000000000000000",
+                    "end": "030000000000000000000000000000000000",
+                },
+                "sub_compaction_max_job_size_mb": 16,
+            },
+        )
+        # sleep random seconds between 0 and max(compaction_time); if the result is 0, wait until the compaction is complete
+        # This would hopefully trigger the restart at different periods of the compaction:
+        # - while we are doing the compaction
+        # - while we finished the compaction but not yet uploaded the metadata
+        # - after we uploaded the metadata
+        time_to_sleep = random.randint(0, max(5, math.ceil(expected_compaction_time_seconds)))
+        if time_to_sleep == 0 or i == 1:
+            start = time.time()
+            wait_until(compaction_finished, timeout=60)
+            end = time.time()
+            expected_compaction_time_seconds = end - start
+            log.info(
+                f"expected_compaction_time_seconds updated to {expected_compaction_time_seconds} seconds"
+            )
+        else:
+            time.sleep(time_to_sleep)
+        env.pageserver.restart(True)
+        ps_http.timeline_gc(
+            tenant_id, timeline_id, None
+        )  # Force refresh gc info to have gc_cutoff generated
+        ps_http.timeline_compact(
+            tenant_id,
+            timeline_id,
+            enhanced_gc_bottom_most_compaction=True,
+            body={
+                "scheduled": True,
+                "sub_compaction": True,
+                "compact_key_range": {
+                    "start": "000000000000000000000000000000000000",
+                    "end": "030000000000000000000000000000000000",
+                },
+                "sub_compaction_max_job_size_mb": 16,
+            },
+        )
+        workload.validate(env.pageserver.id)
+
+    wait_until(compaction_finished, timeout=60)
+
+    # ensure gc_compaction is scheduled and it's actually running (instead of skipping due to no layers picked)
+    env.pageserver.assert_log_contains(
+        "scheduled_compact_timeline.*picked .* layers for compaction"
+    )
+
+    log.info("Validating at workload end ...")
+    workload.validate(env.pageserver.id)
+
+    # Run a legacy compaction+gc to ensure gc-compaction can coexist with legacy compaction.
+    ps_http.timeline_checkpoint(tenant_id, timeline_id, wait_until_uploaded=True)
+    ps_http.timeline_gc(tenant_id, timeline_id, None)
+
+
 # Stripe sizes in number of pages.
 TINY_STRIPES = 16
 LARGE_STRIPES = 32768
@@ -237,7 +463,9 @@ def test_sharding_compaction(
         "pitr_interval": "0s",
         # disable background compaction and GC. We invoke it manually when we want it to happen.
         "gc_period": "0s",
+        "gc_horizon": f"{128 * 1024}",
         "compaction_period": "0s",
+        "lsn_lease_length": "0s",
         # create image layers eagerly: we want to exercise image layer creation in this test.
         "image_creation_threshold": "1",
         "image_layer_creation_check_threshold": 0,
@@ -312,6 +540,8 @@ def test_sharding_compaction(
         for shard in env.storage_controller.locate(tenant_id):
             pageserver = env.get_pageserver(shard["node_id"])
             tenant_shard_id = shard["shard_id"]
+            # Force refresh gc info to have gc_cutoff generated
+            pageserver.http_client().timeline_gc(tenant_shard_id, timeline_id, None)
             pageserver.http_client().timeline_compact(
                 tenant_shard_id,
                 timeline_id,

test_runner/regress/test_compatibility.py

@@ -143,7 +143,7 @@ def test_create_snapshot(
 
     env = neon_env_builder.init_start(
         initial_tenant_conf={
-            # Miniature layers to enable generating non-trivial layer map without writing lots of data
+            # Miniature layers to enable generating non-trivial layer map without writing lots of data.
             "checkpoint_distance": f"{128 * 1024}",
             "compaction_threshold": "1",
             "compaction_target_size": f"{128 * 1024}",

test_runner/regress/test_import_pgdata.py

@@ -1,9 +1,7 @@
-import base64
 import json
 import re
 import time
 from enum import Enum
-from pathlib import Path
 
 import psycopg2
 import psycopg2.errors
@@ -18,11 +16,8 @@ from fixtures.pageserver.http import (
 )
 from fixtures.pg_version import PgVersion
 from fixtures.port_distributor import PortDistributor
-from fixtures.remote_storage import MockS3Server, RemoteStorageKind
+from fixtures.remote_storage import RemoteStorageKind
 from fixtures.utils import run_only_on_postgres
-from mypy_boto3_kms import KMSClient
-from mypy_boto3_kms.type_defs import EncryptResponseTypeDef
-from mypy_boto3_s3 import S3Client
 from pytest_httpserver import HTTPServer
 from werkzeug.wrappers.request import Request
 from werkzeug.wrappers.response import Response
@@ -105,15 +100,13 @@ def test_pgdata_import_smoke(
     while True:
         relblock_size = vanilla_pg.safe_psql_scalar("select pg_relation_size('t')")
         log.info(
-            f"relblock size: {relblock_size / 8192} pages (target: {target_relblock_size // 8192}) pages"
+            f"relblock size: {relblock_size/8192} pages (target: {target_relblock_size//8192}) pages"
         )
         if relblock_size >= target_relblock_size:
             break
         addrows = int((target_relblock_size - relblock_size) // 8192)
         assert addrows >= 1, "forward progress"
-        vanilla_pg.safe_psql(
-            f"insert into t select generate_series({nrows + 1}, {nrows + addrows})"
-        )
+        vanilla_pg.safe_psql(f"insert into t select generate_series({nrows+1}, {nrows + addrows})")
         nrows += addrows
     expect_nrows = nrows
     expect_sum = (
@@ -324,6 +317,10 @@ def test_pgdata_import_smoke(
         br_initdb_endpoint.safe_psql("select * from othertable")
 
 
+@run_only_on_postgres(
+    [PgVersion.V14, PgVersion.V15, PgVersion.V16],
+    "newer control file catalog version and struct format isn't supported",
+)
 def test_fast_import_binary(
     test_output_dir,
     vanilla_pg: VanillaPostgres,
@@ -350,104 +347,6 @@ def test_fast_import_binary(
         assert res[0][0] == 10
 
 
-def test_fast_import_restore_to_connstring(
-    test_output_dir,
-    vanilla_pg: VanillaPostgres,
-    port_distributor: PortDistributor,
-    fast_import: FastImport,
-    pg_distrib_dir: Path,
-    pg_version: PgVersion,
-):
-    vanilla_pg.start()
-    vanilla_pg.safe_psql("CREATE TABLE foo (a int); INSERT INTO foo SELECT generate_series(1, 10);")
-
-    pgdatadir = test_output_dir / "restore-pgdata"
-    pg_bin = PgBin(test_output_dir, pg_distrib_dir, pg_version)
-    port = port_distributor.get_port()
-    with VanillaPostgres(pgdatadir, pg_bin, port) as restore_vanilla_pg:
-        restore_vanilla_pg.configure(["shared_preload_libraries='neon_rmgr'"])
-        restore_vanilla_pg.start()
-
-        fast_import.run(
-            source_connection_string=vanilla_pg.connstr(),
-            restore_connection_string=restore_vanilla_pg.connstr(),
-        )
-        vanilla_pg.stop()
-
-        res = restore_vanilla_pg.safe_psql("SELECT count(*) FROM foo;")
-        log.info(f"Result: {res}")
-        assert res[0][0] == 10
-
-
-def test_fast_import_restore_to_connstring_from_s3_spec(
-    test_output_dir,
-    vanilla_pg: VanillaPostgres,
-    port_distributor: PortDistributor,
-    fast_import: FastImport,
-    pg_distrib_dir: Path,
-    pg_version: PgVersion,
-    mock_s3_server: MockS3Server,
-    mock_kms: KMSClient,
-    mock_s3_client: S3Client,
-):
-    # Prepare KMS and S3
-    key_response = mock_kms.create_key(
-        Description="Test key",
-        KeyUsage="ENCRYPT_DECRYPT",
-        Origin="AWS_KMS",
-    )
-    key_id = key_response["KeyMetadata"]["KeyId"]
-
-    def encrypt(x: str) -> EncryptResponseTypeDef:
-        return mock_kms.encrypt(KeyId=key_id, Plaintext=x)
-
-    # Start source postgres and ingest data
-    vanilla_pg.start()
-    vanilla_pg.safe_psql("CREATE TABLE foo (a int); INSERT INTO foo SELECT generate_series(1, 10);")
-
-    # Start target postgres
-    pgdatadir = test_output_dir / "restore-pgdata"
-    pg_bin = PgBin(test_output_dir, pg_distrib_dir, pg_version)
-    port = port_distributor.get_port()
-    with VanillaPostgres(pgdatadir, pg_bin, port) as restore_vanilla_pg:
-        restore_vanilla_pg.configure(["shared_preload_libraries='neon_rmgr'"])
-        restore_vanilla_pg.start()
-
-        # Encrypt connstrings and put spec into S3
-        source_connstring_encrypted = encrypt(vanilla_pg.connstr())
-        restore_connstring_encrypted = encrypt(restore_vanilla_pg.connstr())
-        spec = {
-            "encryption_secret": {"KMS": {"key_id": key_id}},
-            "source_connstring_ciphertext_base64": base64.b64encode(
-                source_connstring_encrypted["CiphertextBlob"]
-            ).decode("utf-8"),
-            "restore_connstring_ciphertext_base64": base64.b64encode(
-                restore_connstring_encrypted["CiphertextBlob"]
-            ).decode("utf-8"),
-        }
-
-        mock_s3_client.create_bucket(Bucket="test-bucket")
-        mock_s3_client.put_object(
-            Bucket="test-bucket", Key="test-prefix/spec.json", Body=json.dumps(spec)
-        )
-
-        # Run fast_import
-        if fast_import.extra_env is None:
-            fast_import.extra_env = {}
-        fast_import.extra_env["AWS_ACCESS_KEY_ID"] = mock_s3_server.access_key()
-        fast_import.extra_env["AWS_SECRET_ACCESS_KEY"] = mock_s3_server.secret_key()
-        fast_import.extra_env["AWS_SESSION_TOKEN"] = mock_s3_server.session_token()
-        fast_import.extra_env["AWS_REGION"] = mock_s3_server.region()
-        fast_import.extra_env["AWS_ENDPOINT_URL"] = mock_s3_server.endpoint()
-        fast_import.extra_env["RUST_LOG"] = "aws_config=debug,aws_sdk_kms=debug"
-        fast_import.run(s3prefix="s3://test-bucket/test-prefix")
-        vanilla_pg.stop()
-
-        res = restore_vanilla_pg.safe_psql("SELECT count(*) FROM foo;")
-        log.info(f"Result: {res}")
-        assert res[0][0] == 10
-
-
 # TODO: Maybe test with pageserver?
 # 1. run whole neon env
 # 2. create timeline with some s3 path???

test_runner/regress/test_recovery.py

@@ -11,10 +11,13 @@ from fixtures.neon_fixtures import NeonEnvBuilder
 # Test pageserver recovery after crash
 #
 def test_pageserver_recovery(neon_env_builder: NeonEnvBuilder):
-    # Override default checkpointer settings to run it more often
+    # Override default checkpointer settings to run it more often.
+    # This also creates a bunch more L0 layers, so disable backpressure.
     env = neon_env_builder.init_start(
         initial_tenant_conf={
             "checkpoint_distance": "1048576",
+            "l0_flush_delay_threshold": "0",
+            "l0_flush_stall_threshold": "0",
         }
     )
     env.pageserver.is_testing_enabled_or_skip()

test_runner/regress/test_remote_storage.py

@@ -539,6 +539,8 @@ def test_timeline_deletion_with_files_stuck_in_upload_queue(
             # small checkpointing and compaction targets to ensure we generate many operations
             "checkpoint_distance": f"{64 * 1024}",
             "compaction_threshold": "1",
+            "l0_flush_delay_threshold": "0",
+            "l0_flush_stall_threshold": "0",
             "compaction_target_size": f"{64 * 1024}",
             # large horizon to avoid automatic GC (our assert on gc_result below relies on that)
             "gc_horizon": f"{1024 ** 4}",
@@ -784,54 +786,6 @@ def test_empty_branch_remote_storage_upload_on_restart(neon_env_builder: NeonEnv
         create_thread.join()
 
 
-def test_paused_upload_stalls_checkpoint(
-    neon_env_builder: NeonEnvBuilder,
-):
-    """
-    This test checks that checkpoints block on uploads to remote storage.
-    """
-    neon_env_builder.enable_pageserver_remote_storage(RemoteStorageKind.LOCAL_FS)
-
-    env = neon_env_builder.init_start(
-        initial_tenant_conf={
-            # Set a small compaction threshold
-            "compaction_threshold": "3",
-            # Disable GC
-            "gc_period": "0s",
-            # disable PITR
-            "pitr_interval": "0s",
-        }
-    )
-
-    env.pageserver.allowed_errors.append(
-        f".*PUT.* path=/v1/tenant/{env.initial_tenant}/timeline.* request was dropped before completing"
-    )
-
-    tenant_id = env.initial_tenant
-    timeline_id = env.initial_timeline
-
-    client = env.pageserver.http_client()
-    layers_at_creation = client.layer_map_info(tenant_id, timeline_id)
-    deltas_at_creation = len(layers_at_creation.delta_layers())
-    assert (
-        deltas_at_creation == 1
-    ), "are you fixing #5863? make sure we end up with 2 deltas at the end of endpoint lifecycle"
-
-    # Make new layer uploads get stuck.
-    # Note that timeline creation waits for the initial layers to reach remote storage.
-    # So at this point, the `layers_at_creation` are in remote storage.
-    client.configure_failpoints(("before-upload-layer-pausable", "pause"))
-
-    with env.endpoints.create_start("main", tenant_id=tenant_id) as endpoint:
-        # Build two tables with some data inside
-        endpoint.safe_psql("CREATE TABLE foo AS SELECT x FROM generate_series(1, 10000) g(x)")
-        wait_for_last_flush_lsn(env, endpoint, tenant_id, timeline_id)
-
-        with pytest.raises(ReadTimeout):
-            client.timeline_checkpoint(tenant_id, timeline_id, timeout=5)
-        client.configure_failpoints(("before-upload-layer-pausable", "off"))
-
-
 def wait_upload_queue_empty(
     client: PageserverHttpClient, tenant_id: TenantId, timeline_id: TimelineId
 ):

test_runner/regress/test_storage_scrubber.py

@@ -271,8 +271,14 @@ def test_scrubber_physical_gc_ancestors(neon_env_builder: NeonEnvBuilder, shard_
         ps.http_client().timeline_compact(
             shard, timeline_id, force_image_layer_creation=True, wait_until_uploaded=True
         )
-        # Add some WAL so that we don't gc at the latest remote consistent lsn
-        workload.churn_rows(1)
+
+    # Add some WAL so that we don't gc at the latest remote consistent lsn
+    workload.churn_rows(10)
+
+    # Now gc the old stuff away
+    for shard in shards:
+        ps = env.get_tenant_pageserver(shard)
+        assert ps is not None
         ps.http_client().timeline_gc(shard, timeline_id, 0)
 
     # We will use a min_age_secs=1 threshold for deletion, let it pass

test_runner/regress/test_subscriber_branching.py

@@ -0,0 +1,242 @@
+from __future__ import annotations
+
+import time
+
+from fixtures.log_helper import log
+from fixtures.neon_fixtures import NeonEnv, logical_replication_sync
+from fixtures.utils import query_scalar, wait_until
+
+
+# This test checks that branching of timeline with logical subscriptions
+# does not affect logical replication for parent.
+# Endpoint on a new branch will drop all existing subscriptions at the start,
+# so it will not receive any changes.
+# If needed, user can create new subscriptions on the child branch.
+def test_subscriber_branching(neon_simple_env: NeonEnv):
+    env = neon_simple_env
+    env.create_branch("publisher")
+    pub = env.endpoints.create("publisher")
+    pub.respec(
+        skip_pg_catalog_updates=False,
+        create_test_user=True,
+    )
+    pub.start(create_test_user=True)
+
+    env.create_branch("subscriber")
+    sub = env.endpoints.create("subscriber")
+    # Pass create_test_user flag to get properly filled spec.users and spec.databases fields.
+    #
+    # This test checks the per-database operations that happen at compute start
+    # and these operations are applied to the databases that are present in the spec.
+    sub.respec(
+        skip_pg_catalog_updates=False,
+        create_test_user=True,
+    )
+    sub.start(create_test_user=True)
+
+    pub.wait_for_migrations()
+    sub.wait_for_migrations()
+
+    n_records = 1000
+
+    def check_that_changes_propagated():
+        scur.execute("SELECT count(*) FROM t")
+        res = scur.fetchall()
+        assert res[0][0] == n_records
+
+    def insert_data(pub, start):
+        with pub.cursor(dbname="neondb", user="test", password="pubtestpwd") as pcur:
+            for i in range(start, start + n_records):
+                pcur.execute("INSERT into t values (%s,random()*100000)", (i,))
+
+    # create_test_user creates a user without password
+    # but psycopg2 execute() requires a password
+    with sub.cursor() as scur:
+        scur.execute("ALTER USER test WITH PASSWORD 'testpwd'")
+    with pub.cursor() as pcur:
+        # Create a test user to avoid using superuser
+        pcur.execute("ALTER USER test WITH PASSWORD 'pubtestpwd'")
+        # If we don't do this, creating the subscription will fail
+        pub.edit_hba(["host all test 0.0.0.0/0 md5"])
+
+    with pub.cursor(dbname="neondb", user="test", password="pubtestpwd") as pcur:
+        pcur.execute("CREATE TABLE t (pk integer primary key, sk integer)")
+        pcur.execute("CREATE PUBLICATION pub FOR TABLE t")
+
+        with sub.cursor(dbname="neondb", user="test", password="testpwd") as scur:
+            scur.execute("CREATE TABLE t (pk integer primary key, sk integer)")
+            pub_conn = (
+                f"host=localhost port={pub.pg_port} dbname=neondb user=test password=pubtestpwd"
+            )
+            query = f"CREATE SUBSCRIPTION sub CONNECTION '{pub_conn}' PUBLICATION pub"
+            scur.execute(query)
+            time.sleep(2)  # let initial table sync complete
+
+        insert_data(pub, 0)
+
+        with sub.cursor(dbname="neondb", user="test", password="testpwd") as scur:
+            wait_until(check_that_changes_propagated)
+            latest_end_lsn = query_scalar(
+                scur, "select latest_end_lsn from pg_catalog.pg_stat_subscription; "
+            )
+            last_insert_lsn = query_scalar(scur, "select pg_current_wal_insert_lsn();")
+
+            log.info(f"latest_end_lsn = {latest_end_lsn}")
+            log.info(f"last_insert_lsn = {last_insert_lsn}")
+
+        # stop the parent subscriber so that it doesn't interfere with the test
+        sub.stop()
+
+        # 1. good scenario:
+        # create subscriber_child_1
+        # it will not get changes from publisher, because drop_subscriptions_before_start is set to True
+        sub_child_1_timeline_id = env.create_branch(
+            "subscriber_child_1",
+            ancestor_branch_name="subscriber",
+            ancestor_start_lsn=last_insert_lsn,
+        )
+        sub_child_1 = env.endpoints.create("subscriber_child_1")
+        # Pass drop_subscriptions_before_start flag
+        sub_child_1.respec(
+            skip_pg_catalog_updates=False,
+            create_test_user=True,
+            drop_subscriptions_before_start=True,
+        )
+        sub_child_1.start(create_test_user=True)
+
+        # ensure that subscriber_child_1 sees all the data
+        with sub_child_1.cursor(dbname="neondb", user="test", password="testpwd") as scur:
+            scur.execute("SELECT count(*) FROM t")
+            res = scur.fetchall()
+            assert res[0][0] == n_records
+
+            # ensure that there are no subscriptions in this database
+            scur.execute("SELECT 1 FROM pg_catalog.pg_subscription WHERE subname = 'sub'")
+            assert len(scur.fetchall()) == 0
+
+        # ensure that drop_subscriptions_done happened on this timeline
+        with sub_child_1.cursor() as scur_postgres:
+            scur_postgres.execute("SELECT timeline_id from neon.drop_subscriptions_done")
+            res = scur_postgres.fetchall()
+            assert len(res) == 1
+            assert str(sub_child_1_timeline_id) == res[0][0]
+
+        old_n_records = n_records
+        # insert more data on publisher
+        insert_data(pub, n_records)
+        n_records += n_records
+
+        pcur.execute("SELECT count(*) FROM t")
+        res = pcur.fetchall()
+        assert res[0][0] == n_records
+
+        # ensure that subscriber_child_1 doesn't see the new data
+        with sub_child_1.cursor(dbname="neondb", user="test", password="testpwd") as scur:
+            scur.execute("SELECT count(*) FROM t")
+            res = scur.fetchall()
+            assert res[0][0] == old_n_records
+
+        # reenable logical replication on subscriber_child_1
+        # using new publication
+        # ensure that new publication works as expected
+        with sub_child_1.cursor(dbname="neondb", user="test", password="testpwd") as scur:
+            scur.execute("TRUNCATE t")
+
+            # create new subscription
+            # with new pub name
+            pcur.execute("CREATE PUBLICATION pub_new FOR TABLE t")
+            query = f"CREATE SUBSCRIPTION sub_new CONNECTION '{pub_conn}' PUBLICATION pub_new"
+            scur.execute(query)
+
+            wait_until(check_that_changes_propagated)
+
+            scur.execute("SELECT count(*) FROM t")
+            res = scur.fetchall()
+            assert res[0][0] == n_records
+
+        # ensure that new publication works as expected after compute restart
+        # first restart with drop_subscriptions_before_start=True
+        # to emulate the case when compute restarts within the VM with stale spec
+        sub_child_1.stop()
+        sub_child_1.respec(
+            skip_pg_catalog_updates=False,
+            create_test_user=True,
+            drop_subscriptions_before_start=True,
+        )
+        sub_child_1.start(create_test_user=True)
+
+        with sub_child_1.cursor(dbname="neondb", user="test", password="testpwd") as scur:
+            # ensure that even though the flag is set, we didn't drop new subscription
+            scur.execute("SELECT 1 FROM pg_catalog.pg_subscription WHERE subname = 'sub_new'")
+            assert len(scur.fetchall()) == 1
+
+        # ensure that drop_subscriptions_done happened on this timeline
+        with sub_child_1.cursor() as scur_postgres:
+            scur_postgres.execute("SELECT timeline_id from neon.drop_subscriptions_done")
+            res = scur_postgres.fetchall()
+            assert len(res) == 1
+            assert str(sub_child_1_timeline_id) == res[0][0]
+
+        sub_child_1.stop()
+        sub_child_1.respec(
+            skip_pg_catalog_updates=False,
+            create_test_user=True,
+            drop_subscriptions_before_start=False,
+        )
+        sub_child_1.start(create_test_user=True)
+
+        # insert more data on publisher
+        insert_data(pub, n_records)
+        n_records += n_records
+        with sub_child_1.cursor(dbname="neondb", user="test", password="testpwd") as scur:
+            # ensure that there is a subscriptions in this database
+            scur.execute("SELECT 1 FROM pg_catalog.pg_subscription WHERE subname = 'sub_new'")
+            assert len(scur.fetchall()) == 1
+
+            wait_until(check_that_changes_propagated)
+            scur.execute("SELECT count(*) FROM t")
+            res = scur.fetchall()
+            assert res[0][0] == n_records
+
+        # ensure that drop_subscriptions_done happened on this timeline
+        with sub_child_1.cursor() as scur_postgres:
+            scur_postgres.execute("SELECT timeline_id from neon.drop_subscriptions_done")
+            res = scur_postgres.fetchall()
+            assert len(res) == 1
+            assert str(sub_child_1_timeline_id) == res[0][0]
+
+        # wake the sub and ensure that it catches up with the new data
+        sub.start(create_test_user=True)
+        with sub.cursor(dbname="neondb", user="test", password="testpwd") as scur:
+            logical_replication_sync(sub, pub)
+            wait_until(check_that_changes_propagated)
+            scur.execute("SELECT count(*) FROM t")
+            res = scur.fetchall()
+            assert res[0][0] == n_records
+
+        # test that we can create a branch of a branch
+        sub_child_2_timeline_id = env.create_branch(
+            "subscriber_child_2",
+            ancestor_branch_name="subscriber_child_1",
+        )
+        sub_child_2 = env.endpoints.create("subscriber_child_2")
+        # Pass drop_subscriptions_before_start flag
+        sub_child_2.respec(
+            skip_pg_catalog_updates=False,
+            drop_subscriptions_before_start=True,
+        )
+        sub_child_2.start(create_test_user=True)
+
+        # ensure that subscriber_child_2 does not inherit subscription from child_1
+        with sub_child_2.cursor(dbname="neondb", user="test", password="testpwd") as scur:
+            # ensure that there are no subscriptions in this database
+            scur.execute("SELECT count(*) FROM pg_catalog.pg_subscription")
+            res = scur.fetchall()
+            assert res[0][0] == 0
+
+        # ensure that drop_subscriptions_done happened on this timeline
+        with sub_child_2.cursor() as scur_postgres:
+            scur_postgres.execute("SELECT timeline_id from neon.drop_subscriptions_done")
+            res = scur_postgres.fetchall()
+            assert len(res) == 1
+            assert str(sub_child_2_timeline_id) == res[0][0]

test_runner/regress/test_timeline_detach_ancestor.py

@@ -607,7 +607,7 @@ def test_timeline_ancestor_detach_idempotent_success(
 
     if shards_after > 1:
         # FIXME: should this be in the neon_env_builder.init_start?
-        env.storage_controller.reconcile_until_idle()
+        env.storage_controller.reconcile_until_idle(timeout_secs=120)
         client = env.storage_controller.pageserver_api()
     else:
         client = env.pageserver.http_client()
@@ -636,7 +636,7 @@ def test_timeline_ancestor_detach_idempotent_success(
         # Do a shard split
         # This is a reproducer for https://github.com/neondatabase/neon/issues/9667
         env.storage_controller.tenant_shard_split(env.initial_tenant, shards_after)
-        env.storage_controller.reconcile_until_idle()
+        env.storage_controller.reconcile_until_idle(timeout_secs=120)
 
     first_reparenting_response = client.detach_ancestor(env.initial_tenant, first_branch)
     assert set(first_reparenting_response) == {reparented1, reparented2}

test_runner/regress/test_timeline_size.py

@@ -440,7 +440,7 @@ def test_timeline_physical_size_post_compaction(neon_env_builder: NeonEnvBuilder
     env = neon_env_builder.init_start(
         initial_tenant_conf={
             "checkpoint_distance": "100000",
-            "compaction_period": "10m",
+            "compaction_period": "0s",
         }
     )
     pageserver_http = env.pageserver.http_client()

test_runner/regress/test_vm_bits.py

@@ -203,6 +203,9 @@ def test_vm_bit_clear_on_heap_lock_blackbox(neon_env_builder: NeonEnvBuilder):
         "checkpoint_distance": f"{128 * 1024}",
         "compaction_target_size": f"{128 * 1024}",
         "compaction_threshold": "1",
+        # disable L0 backpressure
+        "l0_flush_delay_threshold": "0",
+        "l0_flush_stall_threshold": "0",
         # create image layers eagerly, so that GC can remove some layers
         "image_creation_threshold": "1",
         # set PITR interval to be small, so we can do GC

test_runner/stubs/h2/__init__.pyi

@@ -0,0 +1 @@
+__version__: str

test_runner/stubs/h2/config.pyi

@@ -1,11 +1,12 @@
 from _typeshed import Incomplete
+from typing import Any
 
 class _BooleanConfigOption:
     name: Incomplete
     attr_name: Incomplete
-    def __init__(self, name) -> None: ...
-    def __get__(self, instance, owner): ...
-    def __set__(self, instance, value) -> None: ...
+    def __init__(self, name: str) -> None: ...
+    def __get__(self, instance: Any, owner: Any) -> bool: ...
+    def __set__(self, instance: Any, value: bool) -> None: ...
 
 class DummyLogger:
     def __init__(self, *vargs) -> None: ...
@@ -15,7 +16,7 @@ class DummyLogger:
 class OutputLogger:
     file: Incomplete
     trace_level: Incomplete
-    def __init__(self, file: Incomplete | None = ..., trace_level: bool = ...) -> None: ...
+    def __init__(self, file: Incomplete | None = None, trace_level: bool = False) -> None: ...
     def debug(self, fmtstr, *args) -> None: ...
     def trace(self, fmtstr, *args) -> None: ...
 
@@ -23,20 +24,12 @@ class H2Configuration:
     client_side: Incomplete
     validate_outbound_headers: Incomplete
     normalize_outbound_headers: Incomplete
+    split_outbound_cookies: Incomplete
     validate_inbound_headers: Incomplete
     normalize_inbound_headers: Incomplete
     logger: Incomplete
-    def __init__(
-        self,
-        client_side: bool = ...,
-        header_encoding: Incomplete | None = ...,
-        validate_outbound_headers: bool = ...,
-        normalize_outbound_headers: bool = ...,
-        validate_inbound_headers: bool = ...,
-        normalize_inbound_headers: bool = ...,
-        logger: Incomplete | None = ...,
-    ) -> None: ...
+    def __init__(self, client_side: bool = True, header_encoding: bool | str | None = None, validate_outbound_headers: bool = True, normalize_outbound_headers: bool = True, split_outbound_cookies: bool = False, validate_inbound_headers: bool = True, normalize_inbound_headers: bool = True, logger: DummyLogger | OutputLogger | None = None) -> None: ...
     @property
-    def header_encoding(self): ...
+    def header_encoding(self) -> bool | str | None: ...
     @header_encoding.setter
-    def header_encoding(self, value) -> None: ...
+    def header_encoding(self, value: bool | str | None) -> None: ...

test_runner/stubs/h2/connection.pyi

@@ -1,72 +1,55 @@
-from enum import Enum, IntEnum
-
-from _typeshed import Incomplete
-
 from .config import H2Configuration as H2Configuration
 from .errors import ErrorCodes as ErrorCodes
-from .events import AlternativeServiceAvailable as AlternativeServiceAvailable
-from .events import ConnectionTerminated as ConnectionTerminated
-from .events import PingAckReceived as PingAckReceived
-from .events import PingReceived as PingReceived
-from .events import PriorityUpdated as PriorityUpdated
-from .events import RemoteSettingsChanged as RemoteSettingsChanged
-from .events import SettingsAcknowledged as SettingsAcknowledged
-from .events import UnknownFrameReceived as UnknownFrameReceived
-from .events import WindowUpdated as WindowUpdated
-from .exceptions import DenialOfServiceError as DenialOfServiceError
-from .exceptions import FlowControlError as FlowControlError
-from .exceptions import FrameTooLargeError as FrameTooLargeError
-from .exceptions import NoAvailableStreamIDError as NoAvailableStreamIDError
-from .exceptions import NoSuchStreamError as NoSuchStreamError
-from .exceptions import ProtocolError as ProtocolError
-from .exceptions import RFC1122Error as RFC1122Error
-from .exceptions import StreamClosedError as StreamClosedError
-from .exceptions import StreamIDTooLowError as StreamIDTooLowError
-from .exceptions import TooManyStreamsError as TooManyStreamsError
+from .events import AlternativeServiceAvailable as AlternativeServiceAvailable, ConnectionTerminated as ConnectionTerminated, Event as Event, InformationalResponseReceived as InformationalResponseReceived, PingAckReceived as PingAckReceived, PingReceived as PingReceived, PriorityUpdated as PriorityUpdated, RemoteSettingsChanged as RemoteSettingsChanged, RequestReceived as RequestReceived, ResponseReceived as ResponseReceived, SettingsAcknowledged as SettingsAcknowledged, TrailersReceived as TrailersReceived, UnknownFrameReceived as UnknownFrameReceived, WindowUpdated as WindowUpdated
+from .exceptions import DenialOfServiceError as DenialOfServiceError, FlowControlError as FlowControlError, FrameTooLargeError as FrameTooLargeError, NoAvailableStreamIDError as NoAvailableStreamIDError, NoSuchStreamError as NoSuchStreamError, ProtocolError as ProtocolError, RFC1122Error as RFC1122Error, StreamClosedError as StreamClosedError, StreamIDTooLowError as StreamIDTooLowError, TooManyStreamsError as TooManyStreamsError
 from .frame_buffer import FrameBuffer as FrameBuffer
-from .settings import SettingCodes as SettingCodes
-from .settings import Settings as Settings
-from .stream import H2Stream as H2Stream
-from .stream import StreamClosedBy as StreamClosedBy
-from .utilities import guard_increment_window as guard_increment_window
+from .settings import ChangedSetting as ChangedSetting, SettingCodes as SettingCodes, Settings as Settings
+from .stream import H2Stream as H2Stream, StreamClosedBy as StreamClosedBy
+from .utilities import SizeLimitDict as SizeLimitDict, guard_increment_window as guard_increment_window
 from .windows import WindowManager as WindowManager
+from _typeshed import Incomplete
+from collections.abc import Iterable
+from enum import Enum, IntEnum
+from hpack.struct import Header as Header, HeaderWeaklyTyped as HeaderWeaklyTyped
+from hyperframe.frame import Frame as Frame
+from typing import Any
 
 class ConnectionState(Enum):
-    IDLE: int
-    CLIENT_OPEN: int
-    SERVER_OPEN: int
-    CLOSED: int
+    IDLE = 0
+    CLIENT_OPEN = 1
+    SERVER_OPEN = 2
+    CLOSED = 3
 
 class ConnectionInputs(Enum):
-    SEND_HEADERS: int
-    SEND_PUSH_PROMISE: int
-    SEND_DATA: int
-    SEND_GOAWAY: int
-    SEND_WINDOW_UPDATE: int
-    SEND_PING: int
-    SEND_SETTINGS: int
-    SEND_RST_STREAM: int
-    SEND_PRIORITY: int
-    RECV_HEADERS: int
-    RECV_PUSH_PROMISE: int
-    RECV_DATA: int
-    RECV_GOAWAY: int
-    RECV_WINDOW_UPDATE: int
-    RECV_PING: int
-    RECV_SETTINGS: int
-    RECV_RST_STREAM: int
-    RECV_PRIORITY: int
-    SEND_ALTERNATIVE_SERVICE: int
-    RECV_ALTERNATIVE_SERVICE: int
+    SEND_HEADERS = 0
+    SEND_PUSH_PROMISE = 1
+    SEND_DATA = 2
+    SEND_GOAWAY = 3
+    SEND_WINDOW_UPDATE = 4
+    SEND_PING = 5
+    SEND_SETTINGS = 6
+    SEND_RST_STREAM = 7
+    SEND_PRIORITY = 8
+    RECV_HEADERS = 9
+    RECV_PUSH_PROMISE = 10
+    RECV_DATA = 11
+    RECV_GOAWAY = 12
+    RECV_WINDOW_UPDATE = 13
+    RECV_PING = 14
+    RECV_SETTINGS = 15
+    RECV_RST_STREAM = 16
+    RECV_PRIORITY = 17
+    SEND_ALTERNATIVE_SERVICE = 18
+    RECV_ALTERNATIVE_SERVICE = 19
 
 class AllowedStreamIDs(IntEnum):
-    EVEN: int
-    ODD: int
+    EVEN = 0
+    ODD = 1
 
 class H2ConnectionStateMachine:
     state: Incomplete
     def __init__(self) -> None: ...
-    def process_input(self, input_): ...
+    def process_input(self, input_: ConnectionInputs) -> list[Event]: ...
 
 class H2Connection:
     DEFAULT_MAX_OUTBOUND_FRAME_SIZE: int
@@ -88,55 +71,30 @@ class H2Connection:
     max_outbound_frame_size: Incomplete
     max_inbound_frame_size: Incomplete
     incoming_buffer: Incomplete
-    def __init__(self, config: Incomplete | None = ...) -> None: ...
+    def __init__(self, config: H2Configuration | None = None) -> None: ...
     @property
-    def open_outbound_streams(self): ...
+    def open_outbound_streams(self) -> int: ...
     @property
-    def open_inbound_streams(self): ...
+    def open_inbound_streams(self) -> int: ...
     @property
-    def inbound_flow_control_window(self): ...
+    def inbound_flow_control_window(self) -> int: ...
     def initiate_connection(self) -> None: ...
-    def initiate_upgrade_connection(self, settings_header: Incomplete | None = ...): ...
-    def get_next_available_stream_id(self): ...
-    def send_headers(
-        self,
-        stream_id,
-        headers,
-        end_stream: bool = ...,
-        priority_weight: Incomplete | None = ...,
-        priority_depends_on: Incomplete | None = ...,
-        priority_exclusive: Incomplete | None = ...,
-    ) -> None: ...
-    def send_data(
-        self, stream_id, data, end_stream: bool = ..., pad_length: Incomplete | None = ...
-    ) -> None: ...
-    def end_stream(self, stream_id) -> None: ...
-    def increment_flow_control_window(
-        self, increment, stream_id: Incomplete | None = ...
-    ) -> None: ...
-    def push_stream(self, stream_id, promised_stream_id, request_headers) -> None: ...
-    def ping(self, opaque_data) -> None: ...
-    def reset_stream(self, stream_id, error_code: int = ...) -> None: ...
-    def close_connection(
-        self,
-        error_code: int = ...,
-        additional_data: Incomplete | None = ...,
-        last_stream_id: Incomplete | None = ...,
-    ) -> None: ...
-    def update_settings(self, new_settings) -> None: ...
-    def advertise_alternative_service(
-        self, field_value, origin: Incomplete | None = ..., stream_id: Incomplete | None = ...
-    ) -> None: ...
-    def prioritize(
-        self,
-        stream_id,
-        weight: Incomplete | None = ...,
-        depends_on: Incomplete | None = ...,
-        exclusive: Incomplete | None = ...,
-    ) -> None: ...
-    def local_flow_control_window(self, stream_id): ...
-    def remote_flow_control_window(self, stream_id): ...
-    def acknowledge_received_data(self, acknowledged_size, stream_id) -> None: ...
-    def data_to_send(self, amount: Incomplete | None = ...): ...
+    def initiate_upgrade_connection(self, settings_header: bytes | None = None) -> bytes | None: ...
+    def get_next_available_stream_id(self) -> int: ...
+    def send_headers(self, stream_id: int, headers: Iterable[HeaderWeaklyTyped], end_stream: bool = False, priority_weight: int | None = None, priority_depends_on: int | None = None, priority_exclusive: bool | None = None) -> None: ...
+    def send_data(self, stream_id: int, data: bytes | memoryview, end_stream: bool = False, pad_length: Any = None) -> None: ...
+    def end_stream(self, stream_id: int) -> None: ...
+    def increment_flow_control_window(self, increment: int, stream_id: int | None = None) -> None: ...
+    def push_stream(self, stream_id: int, promised_stream_id: int, request_headers: Iterable[HeaderWeaklyTyped]) -> None: ...
+    def ping(self, opaque_data: bytes | str) -> None: ...
+    def reset_stream(self, stream_id: int, error_code: ErrorCodes | int = 0) -> None: ...
+    def close_connection(self, error_code: ErrorCodes | int = 0, additional_data: bytes | None = None, last_stream_id: int | None = None) -> None: ...
+    def update_settings(self, new_settings: dict[SettingCodes | int, int]) -> None: ...
+    def advertise_alternative_service(self, field_value: bytes | str, origin: bytes | None = None, stream_id: int | None = None) -> None: ...
+    def prioritize(self, stream_id: int, weight: int | None = None, depends_on: int | None = None, exclusive: bool | None = None) -> None: ...
+    def local_flow_control_window(self, stream_id: int) -> int: ...
+    def remote_flow_control_window(self, stream_id: int) -> int: ...
+    def acknowledge_received_data(self, acknowledged_size: int, stream_id: int) -> None: ...
+    def data_to_send(self, amount: int | None = None) -> bytes: ...
     def clear_outbound_data_buffer(self) -> None: ...
-    def receive_data(self, data): ...
+    def receive_data(self, data: bytes) -> list[Event]: ...

test_runner/stubs/h2/errors.pyi

@@ -1,17 +1,19 @@
 import enum
 
+__all__ = ['ErrorCodes']
+
 class ErrorCodes(enum.IntEnum):
-    NO_ERROR: int
-    PROTOCOL_ERROR: int
-    INTERNAL_ERROR: int
-    FLOW_CONTROL_ERROR: int
-    SETTINGS_TIMEOUT: int
-    STREAM_CLOSED: int
-    FRAME_SIZE_ERROR: int
-    REFUSED_STREAM: int
-    CANCEL: int
-    COMPRESSION_ERROR: int
-    CONNECT_ERROR: int
-    ENHANCE_YOUR_CALM: int
-    INADEQUATE_SECURITY: int
-    HTTP_1_1_REQUIRED: int
+    NO_ERROR = 0
+    PROTOCOL_ERROR = 1
+    INTERNAL_ERROR = 2
+    FLOW_CONTROL_ERROR = 3
+    SETTINGS_TIMEOUT = 4
+    STREAM_CLOSED = 5
+    FRAME_SIZE_ERROR = 6
+    REFUSED_STREAM = 7
+    CANCEL = 8
+    COMPRESSION_ERROR = 9
+    CONNECT_ERROR = 10
+    ENHANCE_YOUR_CALM = 11
+    INADEQUATE_SECURITY = 12
+    HTTP_1_1_REQUIRED = 13

test_runner/stubs/h2/events.pyi

@@ -1,6 +1,8 @@
+from .errors import ErrorCodes as ErrorCodes
+from .settings import ChangedSetting as ChangedSetting, SettingCodes as SettingCodes, Settings as Settings
 from _typeshed import Incomplete
-
-from .settings import ChangedSetting as ChangedSetting
+from hpack import HeaderTuple as HeaderTuple
+from hyperframe.frame import Frame as Frame
 
 class Event: ...
 
@@ -53,7 +55,7 @@ class RemoteSettingsChanged(Event):
     changed_settings: Incomplete
     def __init__(self) -> None: ...
     @classmethod
-    def from_settings(cls, old_settings, new_settings): ...
+    def from_settings(cls, old_settings: Settings | dict[int, int], new_settings: dict[int, int]) -> RemoteSettingsChanged: ...
 
 class PingReceived(Event):
     ping_data: Incomplete

test_runner/stubs/h2/exceptions.pyi

@@ -1,3 +1,4 @@
+from .errors import ErrorCodes as ErrorCodes
 from _typeshed import Incomplete
 
 class H2Error(Exception): ...
@@ -19,27 +20,27 @@ class FlowControlError(ProtocolError):
 class StreamIDTooLowError(ProtocolError):
     stream_id: Incomplete
     max_stream_id: Incomplete
-    def __init__(self, stream_id, max_stream_id) -> None: ...
+    def __init__(self, stream_id: int, max_stream_id: int) -> None: ...
 
 class NoAvailableStreamIDError(ProtocolError): ...
 
 class NoSuchStreamError(ProtocolError):
     stream_id: Incomplete
-    def __init__(self, stream_id) -> None: ...
+    def __init__(self, stream_id: int) -> None: ...
 
 class StreamClosedError(NoSuchStreamError):
     stream_id: Incomplete
     error_code: Incomplete
-    def __init__(self, stream_id) -> None: ...
+    def __init__(self, stream_id: int) -> None: ...
 
 class InvalidSettingsValueError(ProtocolError, ValueError):
     error_code: Incomplete
-    def __init__(self, msg, error_code) -> None: ...
+    def __init__(self, msg: str, error_code: ErrorCodes) -> None: ...
 
 class InvalidBodyLengthError(ProtocolError):
     expected_length: Incomplete
     actual_length: Incomplete
-    def __init__(self, expected, actual) -> None: ...
+    def __init__(self, expected: int, actual: int) -> None: ...
 
 class UnsupportedFrameError(ProtocolError): ...
 class RFC1122Error(H2Error): ...

test_runner/stubs/h2/frame_buffer.pyi

@@ -1,19 +1,12 @@
-from .exceptions import (
-    FrameDataMissingError as FrameDataMissingError,
-)
-from .exceptions import (
-    FrameTooLargeError as FrameTooLargeError,
-)
-from .exceptions import (
-    ProtocolError as ProtocolError,
-)
+from .exceptions import FrameDataMissingError as FrameDataMissingError, FrameTooLargeError as FrameTooLargeError, ProtocolError as ProtocolError
+from hyperframe.frame import Frame
 
 CONTINUATION_BACKLOG: int
 
 class FrameBuffer:
     data: bytes
     max_frame_size: int
-    def __init__(self, server: bool = ...) -> None: ...
-    def add_data(self, data) -> None: ...
-    def __iter__(self): ...
-    def __next__(self): ...
+    def __init__(self, server: bool = False) -> None: ...
+    def add_data(self, data: bytes) -> None: ...
+    def __iter__(self) -> FrameBuffer: ...
+    def __next__(self) -> Frame: ...

test_runner/stubs/h2/settings.pyi

@@ -1,61 +1,59 @@
 import enum
-from collections.abc import MutableMapping
-from typing import Any
-
+from .errors import ErrorCodes as ErrorCodes
+from .exceptions import InvalidSettingsValueError as InvalidSettingsValueError
 from _typeshed import Incomplete
-from h2.errors import ErrorCodes as ErrorCodes
-from h2.exceptions import InvalidSettingsValueError as InvalidSettingsValueError
+from collections.abc import Iterator, MutableMapping
 
 class SettingCodes(enum.IntEnum):
-    HEADER_TABLE_SIZE: Incomplete
-    ENABLE_PUSH: Incomplete
-    MAX_CONCURRENT_STREAMS: Incomplete
-    INITIAL_WINDOW_SIZE: Incomplete
-    MAX_FRAME_SIZE: Incomplete
-    MAX_HEADER_LIST_SIZE: Incomplete
-    ENABLE_CONNECT_PROTOCOL: Incomplete
+    HEADER_TABLE_SIZE = ...
+    ENABLE_PUSH = ...
+    MAX_CONCURRENT_STREAMS = ...
+    INITIAL_WINDOW_SIZE = ...
+    MAX_FRAME_SIZE = ...
+    MAX_HEADER_LIST_SIZE = ...
+    ENABLE_CONNECT_PROTOCOL = ...
 
 class ChangedSetting:
     setting: Incomplete
     original_value: Incomplete
     new_value: Incomplete
-    def __init__(self, setting, original_value, new_value) -> None: ...
+    def __init__(self, setting: SettingCodes | int, original_value: int | None, new_value: int) -> None: ...
 
-class Settings(MutableMapping[str, Any]):
-    def __init__(self, client: bool = ..., initial_values: Incomplete | None = ...) -> None: ...
-    def acknowledge(self): ...
+class Settings(MutableMapping[SettingCodes | int, int]):
+    def __init__(self, client: bool = True, initial_values: dict[SettingCodes, int] | None = None) -> None: ...
+    def acknowledge(self) -> dict[SettingCodes | int, ChangedSetting]: ...
     @property
-    def header_table_size(self): ...
+    def header_table_size(self) -> int: ...
     @header_table_size.setter
-    def header_table_size(self, value) -> None: ...
+    def header_table_size(self, value: int) -> None: ...
     @property
-    def enable_push(self): ...
+    def enable_push(self) -> int: ...
     @enable_push.setter
-    def enable_push(self, value) -> None: ...
+    def enable_push(self, value: int) -> None: ...
     @property
-    def initial_window_size(self): ...
+    def initial_window_size(self) -> int: ...
     @initial_window_size.setter
-    def initial_window_size(self, value) -> None: ...
+    def initial_window_size(self, value: int) -> None: ...
     @property
-    def max_frame_size(self): ...
+    def max_frame_size(self) -> int: ...
     @max_frame_size.setter
-    def max_frame_size(self, value) -> None: ...
+    def max_frame_size(self, value: int) -> None: ...
     @property
-    def max_concurrent_streams(self): ...
+    def max_concurrent_streams(self) -> int: ...
     @max_concurrent_streams.setter
-    def max_concurrent_streams(self, value) -> None: ...
+    def max_concurrent_streams(self, value: int) -> None: ...
     @property
-    def max_header_list_size(self): ...
+    def max_header_list_size(self) -> int | None: ...
     @max_header_list_size.setter
-    def max_header_list_size(self, value) -> None: ...
+    def max_header_list_size(self, value: int) -> None: ...
     @property
-    def enable_connect_protocol(self): ...
+    def enable_connect_protocol(self) -> int: ...
     @enable_connect_protocol.setter
-    def enable_connect_protocol(self, value) -> None: ...
-    def __getitem__(self, key): ...
-    def __setitem__(self, key, value) -> None: ...
-    def __delitem__(self, key) -> None: ...
-    def __iter__(self): ...
+    def enable_connect_protocol(self, value: int) -> None: ...
+    def __getitem__(self, key: SettingCodes | int) -> int: ...
+    def __setitem__(self, key: SettingCodes | int, value: int) -> None: ...
+    def __delitem__(self, key: SettingCodes | int) -> None: ...
+    def __iter__(self) -> Iterator[SettingCodes | int]: ...
     def __len__(self) -> int: ...
-    def __eq__(self, other): ...
-    def __ne__(self, other): ...
+    def __eq__(self, other: object) -> bool: ...
+    def __ne__(self, other: object) -> bool: ...

test_runner/stubs/h2/stream.pyi

@@ -1,114 +1,52 @@
-from enum import Enum, IntEnum
-
-from _typeshed import Incomplete
-
+from .config import H2Configuration as H2Configuration
 from .errors import ErrorCodes as ErrorCodes
-from .events import (
-    AlternativeServiceAvailable as AlternativeServiceAvailable,
-)
-from .events import (
-    DataReceived as DataReceived,
-)
-from .events import (
-    InformationalResponseReceived as InformationalResponseReceived,
-)
-from .events import (
-    PushedStreamReceived as PushedStreamReceived,
-)
-from .events import (
-    RequestReceived as RequestReceived,
-)
-from .events import (
-    ResponseReceived as ResponseReceived,
-)
-from .events import (
-    StreamEnded as StreamEnded,
-)
-from .events import (
-    StreamReset as StreamReset,
-)
-from .events import (
-    TrailersReceived as TrailersReceived,
-)
-from .events import (
-    WindowUpdated as WindowUpdated,
-)
-from .exceptions import (
-    FlowControlError as FlowControlError,
-)
-from .exceptions import (
-    InvalidBodyLengthError as InvalidBodyLengthError,
-)
-from .exceptions import (
-    ProtocolError as ProtocolError,
-)
-from .exceptions import (
-    StreamClosedError as StreamClosedError,
-)
-from .utilities import (
-    HeaderValidationFlags as HeaderValidationFlags,
-)
-from .utilities import (
-    authority_from_headers as authority_from_headers,
-)
-from .utilities import (
-    extract_method_header as extract_method_header,
-)
-from .utilities import (
-    guard_increment_window as guard_increment_window,
-)
-from .utilities import (
-    is_informational_response as is_informational_response,
-)
-from .utilities import (
-    normalize_inbound_headers as normalize_inbound_headers,
-)
-from .utilities import (
-    normalize_outbound_headers as normalize_outbound_headers,
-)
-from .utilities import (
-    validate_headers as validate_headers,
-)
-from .utilities import (
-    validate_outbound_headers as validate_outbound_headers,
-)
+from .events import AlternativeServiceAvailable as AlternativeServiceAvailable, DataReceived as DataReceived, Event as Event, InformationalResponseReceived as InformationalResponseReceived, PushedStreamReceived as PushedStreamReceived, RequestReceived as RequestReceived, ResponseReceived as ResponseReceived, StreamEnded as StreamEnded, StreamReset as StreamReset, TrailersReceived as TrailersReceived, WindowUpdated as WindowUpdated
+from .exceptions import FlowControlError as FlowControlError, InvalidBodyLengthError as InvalidBodyLengthError, ProtocolError as ProtocolError, StreamClosedError as StreamClosedError
+from .utilities import HeaderValidationFlags as HeaderValidationFlags, authority_from_headers as authority_from_headers, extract_method_header as extract_method_header, guard_increment_window as guard_increment_window, is_informational_response as is_informational_response, normalize_inbound_headers as normalize_inbound_headers, normalize_outbound_headers as normalize_outbound_headers, utf8_encode_headers as utf8_encode_headers, validate_headers as validate_headers, validate_outbound_headers as validate_outbound_headers
 from .windows import WindowManager as WindowManager
+from _typeshed import Incomplete
+from collections.abc import Iterable
+from enum import Enum, IntEnum
+from hpack.hpack import Encoder as Encoder
+from hpack.struct import Header as Header, HeaderWeaklyTyped as HeaderWeaklyTyped
+from hyperframe.frame import AltSvcFrame, ContinuationFrame, Frame as Frame, HeadersFrame, PushPromiseFrame, RstStreamFrame
+from typing import Any
 
 class StreamState(IntEnum):
-    IDLE: int
-    RESERVED_REMOTE: int
-    RESERVED_LOCAL: int
-    OPEN: int
-    HALF_CLOSED_REMOTE: int
-    HALF_CLOSED_LOCAL: int
-    CLOSED: int
+    IDLE = 0
+    RESERVED_REMOTE = 1
+    RESERVED_LOCAL = 2
+    OPEN = 3
+    HALF_CLOSED_REMOTE = 4
+    HALF_CLOSED_LOCAL = 5
+    CLOSED = 6
 
 class StreamInputs(Enum):
-    SEND_HEADERS: int
-    SEND_PUSH_PROMISE: int
-    SEND_RST_STREAM: int
-    SEND_DATA: int
-    SEND_WINDOW_UPDATE: int
-    SEND_END_STREAM: int
-    RECV_HEADERS: int
-    RECV_PUSH_PROMISE: int
-    RECV_RST_STREAM: int
-    RECV_DATA: int
-    RECV_WINDOW_UPDATE: int
-    RECV_END_STREAM: int
-    RECV_CONTINUATION: int
-    SEND_INFORMATIONAL_HEADERS: int
-    RECV_INFORMATIONAL_HEADERS: int
-    SEND_ALTERNATIVE_SERVICE: int
-    RECV_ALTERNATIVE_SERVICE: int
-    UPGRADE_CLIENT: int
-    UPGRADE_SERVER: int
+    SEND_HEADERS = 0
+    SEND_PUSH_PROMISE = 1
+    SEND_RST_STREAM = 2
+    SEND_DATA = 3
+    SEND_WINDOW_UPDATE = 4
+    SEND_END_STREAM = 5
+    RECV_HEADERS = 6
+    RECV_PUSH_PROMISE = 7
+    RECV_RST_STREAM = 8
+    RECV_DATA = 9
+    RECV_WINDOW_UPDATE = 10
+    RECV_END_STREAM = 11
+    RECV_CONTINUATION = 12
+    SEND_INFORMATIONAL_HEADERS = 13
+    RECV_INFORMATIONAL_HEADERS = 14
+    SEND_ALTERNATIVE_SERVICE = 15
+    RECV_ALTERNATIVE_SERVICE = 16
+    UPGRADE_CLIENT = 17
+    UPGRADE_SERVER = 18
 
 class StreamClosedBy(Enum):
-    SEND_END_STREAM: int
-    RECV_END_STREAM: int
-    SEND_RST_STREAM: int
-    RECV_RST_STREAM: int
+    SEND_END_STREAM = 0
+    RECV_END_STREAM = 1
+    SEND_RST_STREAM = 2
+    RECV_RST_STREAM = 3
 
 STREAM_OPEN: Incomplete
 
@@ -121,32 +59,32 @@ class H2StreamStateMachine:
     headers_received: Incomplete
     trailers_received: Incomplete
     stream_closed_by: Incomplete
-    def __init__(self, stream_id) -> None: ...
-    def process_input(self, input_): ...
-    def request_sent(self, previous_state): ...
-    def response_sent(self, previous_state): ...
-    def request_received(self, previous_state): ...
-    def response_received(self, previous_state): ...
-    def data_received(self, previous_state): ...
-    def window_updated(self, previous_state): ...
-    def stream_half_closed(self, previous_state): ...
-    def stream_ended(self, previous_state): ...
-    def stream_reset(self, previous_state): ...
-    def send_new_pushed_stream(self, previous_state): ...
-    def recv_new_pushed_stream(self, previous_state): ...
-    def send_push_promise(self, previous_state): ...
-    def recv_push_promise(self, previous_state): ...
-    def send_end_stream(self, previous_state) -> None: ...
-    def send_reset_stream(self, previous_state) -> None: ...
-    def reset_stream_on_error(self, previous_state) -> None: ...
-    def recv_on_closed_stream(self, previous_state) -> None: ...
-    def send_on_closed_stream(self, previous_state) -> None: ...
-    def recv_push_on_closed_stream(self, previous_state) -> None: ...
-    def send_push_on_closed_stream(self, previous_state) -> None: ...
-    def send_informational_response(self, previous_state): ...
-    def recv_informational_response(self, previous_state): ...
-    def recv_alt_svc(self, previous_state): ...
-    def send_alt_svc(self, previous_state) -> None: ...
+    def __init__(self, stream_id: int) -> None: ...
+    def process_input(self, input_: StreamInputs) -> Any: ...
+    def request_sent(self, previous_state: StreamState) -> list[Event]: ...
+    def response_sent(self, previous_state: StreamState) -> list[Event]: ...
+    def request_received(self, previous_state: StreamState) -> list[Event]: ...
+    def response_received(self, previous_state: StreamState) -> list[Event]: ...
+    def data_received(self, previous_state: StreamState) -> list[Event]: ...
+    def window_updated(self, previous_state: StreamState) -> list[Event]: ...
+    def stream_half_closed(self, previous_state: StreamState) -> list[Event]: ...
+    def stream_ended(self, previous_state: StreamState) -> list[Event]: ...
+    def stream_reset(self, previous_state: StreamState) -> list[Event]: ...
+    def send_new_pushed_stream(self, previous_state: StreamState) -> list[Event]: ...
+    def recv_new_pushed_stream(self, previous_state: StreamState) -> list[Event]: ...
+    def send_push_promise(self, previous_state: StreamState) -> list[Event]: ...
+    def recv_push_promise(self, previous_state: StreamState) -> list[Event]: ...
+    def send_end_stream(self, previous_state: StreamState) -> None: ...
+    def send_reset_stream(self, previous_state: StreamState) -> None: ...
+    def reset_stream_on_error(self, previous_state: StreamState) -> None: ...
+    def recv_on_closed_stream(self, previous_state: StreamState) -> None: ...
+    def send_on_closed_stream(self, previous_state: StreamState) -> None: ...
+    def recv_push_on_closed_stream(self, previous_state: StreamState) -> None: ...
+    def send_push_on_closed_stream(self, previous_state: StreamState) -> None: ...
+    def send_informational_response(self, previous_state: StreamState) -> list[Event]: ...
+    def recv_informational_response(self, previous_state: StreamState) -> list[Event]: ...
+    def recv_alt_svc(self, previous_state: StreamState) -> list[Event]: ...
+    def send_alt_svc(self, previous_state: StreamState) -> None: ...
 
 class H2Stream:
     state_machine: Incomplete
@@ -155,30 +93,30 @@ class H2Stream:
     request_method: Incomplete
     outbound_flow_control_window: Incomplete
     config: Incomplete
-    def __init__(self, stream_id, config, inbound_window_size, outbound_window_size) -> None: ...
+    def __init__(self, stream_id: int, config: H2Configuration, inbound_window_size: int, outbound_window_size: int) -> None: ...
     @property
-    def inbound_flow_control_window(self): ...
+    def inbound_flow_control_window(self) -> int: ...
     @property
-    def open(self): ...
+    def open(self) -> bool: ...
     @property
-    def closed(self): ...
+    def closed(self) -> bool: ...
     @property
-    def closed_by(self): ...
-    def upgrade(self, client_side) -> None: ...
-    def send_headers(self, headers, encoder, end_stream: bool = ...): ...
-    def push_stream_in_band(self, related_stream_id, headers, encoder): ...
-    def locally_pushed(self): ...
-    def send_data(self, data, end_stream: bool = ..., pad_length: Incomplete | None = ...): ...
-    def end_stream(self): ...
-    def advertise_alternative_service(self, field_value): ...
-    def increase_flow_control_window(self, increment): ...
-    def receive_push_promise_in_band(self, promised_stream_id, headers, header_encoding): ...
-    def remotely_pushed(self, pushed_headers): ...
-    def receive_headers(self, headers, end_stream, header_encoding): ...
-    def receive_data(self, data, end_stream, flow_control_len): ...
-    def receive_window_update(self, increment): ...
+    def closed_by(self) -> StreamClosedBy | None: ...
+    def upgrade(self, client_side: bool) -> None: ...
+    def send_headers(self, headers: Iterable[HeaderWeaklyTyped], encoder: Encoder, end_stream: bool = False) -> list[HeadersFrame | ContinuationFrame | PushPromiseFrame]: ...
+    def push_stream_in_band(self, related_stream_id: int, headers: Iterable[HeaderWeaklyTyped], encoder: Encoder) -> list[HeadersFrame | ContinuationFrame | PushPromiseFrame]: ...
+    def locally_pushed(self) -> list[Frame]: ...
+    def send_data(self, data: bytes | memoryview, end_stream: bool = False, pad_length: int | None = None) -> list[Frame]: ...
+    def end_stream(self) -> list[Frame]: ...
+    def advertise_alternative_service(self, field_value: bytes) -> list[Frame]: ...
+    def increase_flow_control_window(self, increment: int) -> list[Frame]: ...
+    def receive_push_promise_in_band(self, promised_stream_id: int, headers: Iterable[Header], header_encoding: bool | str | None) -> tuple[list[Frame], list[Event]]: ...
+    def remotely_pushed(self, pushed_headers: Iterable[Header]) -> tuple[list[Frame], list[Event]]: ...
+    def receive_headers(self, headers: Iterable[Header], end_stream: bool, header_encoding: bool | str | None) -> tuple[list[Frame], list[Event]]: ...
+    def receive_data(self, data: bytes, end_stream: bool, flow_control_len: int) -> tuple[list[Frame], list[Event]]: ...
+    def receive_window_update(self, increment: int) -> tuple[list[Frame], list[Event]]: ...
     def receive_continuation(self) -> None: ...
-    def receive_alt_svc(self, frame): ...
-    def reset_stream(self, error_code: int = ...): ...
-    def stream_reset(self, frame): ...
-    def acknowledge_received_data(self, acknowledged_size): ...
+    def receive_alt_svc(self, frame: AltSvcFrame) -> tuple[list[Frame], list[Event]]: ...
+    def reset_stream(self, error_code: ErrorCodes | int = 0) -> list[Frame]: ...
+    def stream_reset(self, frame: RstStreamFrame) -> tuple[list[Frame], list[Event]]: ...
+    def acknowledge_received_data(self, acknowledged_size: int) -> list[Frame]: ...

test_runner/stubs/h2/utilities.pyi

@@ -1,25 +1,32 @@
-from typing import NamedTuple
-
+import collections
+from .exceptions import FlowControlError as FlowControlError, ProtocolError as ProtocolError
 from _typeshed import Incomplete
-
-from .exceptions import FlowControlError as FlowControlError
-from .exceptions import ProtocolError as ProtocolError
+from collections.abc import Generator, Iterable
+from hpack.struct import Header as Header, HeaderWeaklyTyped as HeaderWeaklyTyped
+from typing import Any, NamedTuple
 
 UPPER_RE: Incomplete
+SIGIL: Incomplete
+INFORMATIONAL_START: Incomplete
 CONNECTION_HEADERS: Incomplete
 
-def extract_method_header(headers): ...
-def is_informational_response(headers): ...
-def guard_increment_window(current, increment): ...
-def authority_from_headers(headers): ...
+def extract_method_header(headers: Iterable[Header]) -> bytes | None: ...
+def is_informational_response(headers: Iterable[Header]) -> bool: ...
+def guard_increment_window(current: int, increment: int) -> int: ...
+def authority_from_headers(headers: Iterable[Header]) -> bytes | None: ...
 
 class HeaderValidationFlags(NamedTuple):
-    is_client: Incomplete
-    is_trailer: Incomplete
-    is_response_header: Incomplete
-    is_push_promise: Incomplete
+    is_client: bool
+    is_trailer: bool
+    is_response_header: bool
+    is_push_promise: bool
 
-def validate_headers(headers, hdr_validation_flags): ...
-def normalize_outbound_headers(headers, hdr_validation_flags): ...
-def normalize_inbound_headers(headers, hdr_validation_flags): ...
-def validate_outbound_headers(headers, hdr_validation_flags): ...
+def validate_headers(headers: Iterable[Header], hdr_validation_flags: HeaderValidationFlags) -> Iterable[Header]: ...
+def utf8_encode_headers(headers: Iterable[HeaderWeaklyTyped]) -> list[Header]: ...
+def normalize_outbound_headers(headers: Iterable[Header], hdr_validation_flags: HeaderValidationFlags | None, should_split_outbound_cookies: bool = False) -> Generator[Header, None, None]: ...
+def normalize_inbound_headers(headers: Iterable[Header], hdr_validation_flags: HeaderValidationFlags) -> Generator[Header, None, None]: ...
+def validate_outbound_headers(headers: Iterable[Header], hdr_validation_flags: HeaderValidationFlags) -> Generator[Header, None, None]: ...
+
+class SizeLimitDict(collections.OrderedDict[int, Any]):
+    def __init__(self, *args: dict[int, int], **kwargs: Any) -> None: ...
+    def __setitem__(self, key: int, value: Any | int) -> None: ...

test_runner/stubs/h2/windows.pyi

@@ -1,13 +1,12 @@
-from _typeshed import Incomplete
-
 from .exceptions import FlowControlError as FlowControlError
+from _typeshed import Incomplete
 
 LARGEST_FLOW_CONTROL_WINDOW: Incomplete
 
 class WindowManager:
     max_window_size: Incomplete
     current_window_size: Incomplete
-    def __init__(self, max_window_size) -> None: ...
-    def window_consumed(self, size) -> None: ...
-    def window_opened(self, size) -> None: ...
-    def process_bytes(self, size): ...
+    def __init__(self, max_window_size: int) -> None: ...
+    def window_consumed(self, size: int) -> None: ...
+    def window_opened(self, size: int) -> None: ...
+    def process_bytes(self, size: int) -> int | None: ...

vendor/revisions.json

@@ -1,18 +1,18 @@
 {
   "v17": [
     "17.2",
-    "a8dd6e779dde907778006adb436b557ad652fb97"
+    "46f9b96555e084c35dd975da9485996db9e86181"
   ],
   "v16": [
     "16.6",
-    "d674efd776f59d78e8fa1535bd2f95c3e6984fca"
+    "3cf7ce1afab75027716d14223f95ddb300754162"
   ],
   "v15": [
     "15.10",
-    "dd0b28d6fbad39e227f3b77296fcca879af8b3a9"
+    "355a7c69d3f907f3612eb406cc7b9c2f55d59b59"
   ],
   "v14": [
     "14.15",
-    "46082f20884f087a2d974b33ac65d63af26142bd"
+    "c0aedfd3cac447510a2db843b561f0c52901b679"
   ]
 }