Initial implementation of GCS provider.

.dockerignore

@@ -19,7 +19,7 @@
 !pageserver/
 !pgxn/
 !proxy/
-!endpoint_storage/
+!object_storage/
 !storage_scrubber/
 !safekeeper/
 !storage_broker/

.github/actions/run-python-test-set/action.yml

@@ -133,7 +133,6 @@ runs:
         fi
 
         PERF_REPORT_DIR="$(realpath test_runner/perf-report-local)"
-        echo "PERF_REPORT_DIR=${PERF_REPORT_DIR}" >> ${GITHUB_ENV}
         rm -rf $PERF_REPORT_DIR
 
         TEST_SELECTION="test_runner/${{ inputs.test_selection }}"
@@ -210,12 +209,11 @@ runs:
           --verbose \
           -rA $TEST_SELECTION $EXTRA_PARAMS
 
-    - name: Upload performance report
-      if: ${{ !cancelled() && inputs.save_perf_report == 'true' }}
-      shell: bash -euxo pipefail {0}
-      run: |
-        export REPORT_FROM="${PERF_REPORT_DIR}"
-        scripts/generate_and_push_perf_report.sh
+        if [[ "${{ inputs.save_perf_report }}" == "true" ]]; then
+          export REPORT_FROM="$PERF_REPORT_DIR"
+          export REPORT_TO="$PLATFORM"
+          scripts/generate_and_push_perf_report.sh
+        fi
 
     - name: Upload compatibility snapshot
       # Note, that we use `github.base_ref` which is a target branch for a PR

Cargo.lock

@@ -2037,33 +2037,6 @@ dependencies = [
  "zeroize",
 ]
 
-[[package]]
-name = "endpoint_storage"
-version = "0.0.1"
-dependencies = [
- "anyhow",
- "axum",
- "axum-extra",
- "camino",
- "camino-tempfile",
- "futures",
- "http-body-util",
- "itertools 0.10.5",
- "jsonwebtoken",
- "prometheus",
- "rand 0.8.5",
- "remote_storage",
- "serde",
- "serde_json",
- "test-log",
- "tokio",
- "tokio-util",
- "tower 0.5.2",
- "tracing",
- "utils",
- "workspace_hack",
-]
-
 [[package]]
 name = "enum-map"
 version = "2.5.0"
@@ -2451,6 +2424,33 @@ dependencies = [
  "slab",
 ]
 
+[[package]]
+name = "gcp_auth"
+version = "0.12.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dbf67f30198e045a039264c01fb44659ce82402d7771c50938beb41a5ac87733"
+dependencies = [
+ "async-trait",
+ "base64 0.22.1",
+ "bytes",
+ "chrono",
+ "home",
+ "http 1.1.0",
+ "http-body-util",
+ "hyper 1.4.1",
+ "hyper-rustls 0.27.5",
+ "hyper-util",
+ "ring",
+ "rustls-pemfile 2.1.1",
+ "serde",
+ "serde_json",
+ "thiserror 1.0.69",
+ "tokio",
+ "tracing",
+ "tracing-futures",
+ "url",
+]
+
 [[package]]
 name = "gen_ops"
 version = "0.4.0"
@@ -2749,6 +2749,15 @@ dependencies = [
  "digest",
 ]
 
+[[package]]
+name = "home"
+version = "0.5.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "589533453244b0995c858700322199b2becb13b627df2851f64a2775d024abcf"
+dependencies = [
+ "windows-sys 0.59.0",
+]
+
 [[package]]
 name = "hostname"
 version = "0.4.0"
@@ -2978,6 +2987,24 @@ dependencies = [
  "tower-service",
 ]
 
+[[package]]
+name = "hyper-rustls"
+version = "0.27.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2d191583f3da1305256f22463b9bb0471acad48a4e534a5218b9963e9c1f59b2"
+dependencies = [
+ "futures-util",
+ "http 1.1.0",
+ "hyper 1.4.1",
+ "hyper-util",
+ "rustls 0.23.18",
+ "rustls-native-certs 0.8.0",
+ "rustls-pki-types",
+ "tokio",
+ "tokio-rustls 0.26.0",
+ "tower-service",
+]
+
 [[package]]
 name = "hyper-timeout"
 version = "0.5.1"
@@ -3733,6 +3760,16 @@ version = "0.3.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a"
 
+[[package]]
+name = "mime_guess"
+version = "2.0.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f7c44f8e672c00fe5308fa235f821cb4198414e1c77935c1ab6948d3fd78550e"
+dependencies = [
+ "mime",
+ "unicase",
+]
+
 [[package]]
 name = "minimal-lexical"
 version = "0.2.1"
@@ -4025,6 +4062,33 @@ dependencies = [
  "memchr",
 ]
 
+[[package]]
+name = "object_storage"
+version = "0.0.1"
+dependencies = [
+ "anyhow",
+ "axum",
+ "axum-extra",
+ "camino",
+ "camino-tempfile",
+ "futures",
+ "http-body-util",
+ "itertools 0.10.5",
+ "jsonwebtoken",
+ "prometheus",
+ "rand 0.8.5",
+ "remote_storage",
+ "serde",
+ "serde_json",
+ "test-log",
+ "tokio",
+ "tokio-util",
+ "tower 0.5.2",
+ "tracing",
+ "utils",
+ "workspace_hack",
+]
+
 [[package]]
 name = "once_cell"
 version = "1.20.2"
@@ -5522,8 +5586,11 @@ dependencies = [
  "bytes",
  "camino",
  "camino-tempfile",
+ "chrono",
  "futures",
  "futures-util",
+ "gcp_auth",
+ "http 1.1.0",
  "http-body-util",
  "http-types",
  "humantime-serde",
@@ -5544,7 +5611,9 @@ dependencies = [
  "tokio-util",
  "toml_edit",
  "tracing",
+ "url",
  "utils",
+ "uuid",
 ]
 
 [[package]]
@@ -5574,6 +5643,7 @@ dependencies = [
  "js-sys",
  "log",
  "mime",
+ "mime_guess",
  "once_cell",
  "percent-encoding",
  "pin-project-lite",
@@ -7597,6 +7667,16 @@ dependencies = [
  "tracing-subscriber",
 ]
 
+[[package]]
+name = "tracing-futures"
+version = "0.2.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "97d095ae15e245a057c8e8451bab9b3ee1e1f68e9ba2b4fbc18d0ac5237835f2"
+dependencies = [
+ "pin-project",
+ "tracing",
+]
+
 [[package]]
 name = "tracing-log"
 version = "0.2.0"
@@ -7750,6 +7830,12 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "unicase"
+version = "2.8.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "75b844d17643ee918803943289730bec8aac480150456169e647ed0b576ba539"
+
 [[package]]
 name = "unicode-bidi"
 version = "0.3.17"

Cargo.toml

@@ -40,7 +40,7 @@ members = [
     "libs/proxy/postgres-protocol2",
     "libs/proxy/postgres-types2",
     "libs/proxy/tokio-postgres2",
-    "endpoint_storage",
+    "object_storage",
 ]
 
 [workspace.package]

Dockerfile

@@ -89,7 +89,7 @@ RUN set -e \
       --bin storage_broker  \
       --bin storage_controller  \
       --bin proxy  \
-      --bin endpoint_storage \
+      --bin object_storage \
       --bin neon_local \
       --bin storage_scrubber \
       --locked --release
@@ -122,7 +122,7 @@ COPY --from=build --chown=neon:neon /home/nonroot/target/release/safekeeper
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/storage_broker      /usr/local/bin
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/storage_controller  /usr/local/bin
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/proxy               /usr/local/bin
-COPY --from=build --chown=neon:neon /home/nonroot/target/release/endpoint_storage    /usr/local/bin
+COPY --from=build --chown=neon:neon /home/nonroot/target/release/object_storage      /usr/local/bin
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/neon_local          /usr/local/bin
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/storage_scrubber    /usr/local/bin
 

compute/compute-node.Dockerfile

@@ -1677,7 +1677,7 @@ RUN set -e \
     && apt clean && rm -rf /var/lib/apt/lists/*
 
 # Use `dist_man_MANS=` to skip manpage generation (which requires python3/pandoc)
-ENV PGBOUNCER_TAG=pgbouncer_1_24_1
+ENV PGBOUNCER_TAG=pgbouncer_1_22_1
 RUN set -e \
     && git clone --recurse-submodules --depth 1 --branch ${PGBOUNCER_TAG} https://github.com/pgbouncer/pgbouncer.git pgbouncer \
     && cd pgbouncer \

compute/patches/pg_repack.patch

@@ -11,14 +11,6 @@ index bf6edcb..89b4c7f 100644
  
  USE_PGXS = 1	# use pgxs if not in contrib directory
  PGXS := $(shell $(PG_CONFIG) --pgxs)
-diff --git a/regress/expected/init-extension.out b/regress/expected/init-extension.out
-index 9f2e171..f6e4f8d 100644
---- a/regress/expected/init-extension.out
-+++ b/regress/expected/init-extension.out
-@@ -1,3 +1,2 @@
- SET client_min_messages = warning;
- CREATE EXTENSION pg_repack;
--RESET client_min_messages;
 diff --git a/regress/expected/nosuper.out b/regress/expected/nosuper.out
 index 8d0a94e..63b68bf 100644
 --- a/regress/expected/nosuper.out
@@ -50,14 +42,6 @@ index 8d0a94e..63b68bf 100644
  INFO: repacking table "public.tbl_cluster"
  ERROR: query failed: ERROR:  current transaction is aborted, commands ignored until end of transaction block
  DETAIL: query was: RESET lock_timeout
-diff --git a/regress/sql/init-extension.sql b/regress/sql/init-extension.sql
-index 9f2e171..f6e4f8d 100644
---- a/regress/sql/init-extension.sql
-+++ b/regress/sql/init-extension.sql
-@@ -1,3 +1,2 @@
- SET client_min_messages = warning;
- CREATE EXTENSION pg_repack;
--RESET client_min_messages;
 diff --git a/regress/sql/nosuper.sql b/regress/sql/nosuper.sql
 index 072f0fa..dbe60f8 100644
 --- a/regress/sql/nosuper.sql

control_plane/src/bin/neon_local.rs

@@ -18,11 +18,12 @@ use anyhow::{Context, Result, anyhow, bail};
 use clap::Parser;
 use compute_api::spec::ComputeMode;
 use control_plane::endpoint::ComputeControlPlane;
-use control_plane::endpoint_storage::{ENDPOINT_STORAGE_DEFAULT_PORT, EndpointStorage};
 use control_plane::local_env::{
-    EndpointStorageConf, InitForceMode, LocalEnv, NeonBroker, NeonLocalInitConf,
-    NeonLocalInitPageserverConf, SafekeeperConf,
+    InitForceMode, LocalEnv, NeonBroker, NeonLocalInitConf, NeonLocalInitPageserverConf,
+    ObjectStorageConf, SafekeeperConf,
 };
+use control_plane::object_storage::OBJECT_STORAGE_DEFAULT_PORT;
+use control_plane::object_storage::ObjectStorage;
 use control_plane::pageserver::PageServerNode;
 use control_plane::safekeeper::SafekeeperNode;
 use control_plane::storage_controller::{
@@ -92,7 +93,7 @@ enum NeonLocalCmd {
     #[command(subcommand)]
     Safekeeper(SafekeeperCmd),
     #[command(subcommand)]
-    EndpointStorage(EndpointStorageCmd),
+    ObjectStorage(ObjectStorageCmd),
     #[command(subcommand)]
     Endpoint(EndpointCmd),
     #[command(subcommand)]
@@ -459,14 +460,14 @@ enum SafekeeperCmd {
 
 #[derive(clap::Subcommand)]
 #[clap(about = "Manage object storage")]
-enum EndpointStorageCmd {
-    Start(EndpointStorageStartCmd),
-    Stop(EndpointStorageStopCmd),
+enum ObjectStorageCmd {
+    Start(ObjectStorageStartCmd),
+    Stop(ObjectStorageStopCmd),
 }
 
 #[derive(clap::Args)]
 #[clap(about = "Start object storage")]
-struct EndpointStorageStartCmd {
+struct ObjectStorageStartCmd {
     #[clap(short = 't', long, help = "timeout until we fail the command")]
     #[arg(default_value = "10s")]
     start_timeout: humantime::Duration,
@@ -474,7 +475,7 @@ struct EndpointStorageStartCmd {
 
 #[derive(clap::Args)]
 #[clap(about = "Stop object storage")]
-struct EndpointStorageStopCmd {
+struct ObjectStorageStopCmd {
     #[arg(value_enum, default_value = "fast")]
     #[clap(
         short = 'm',
@@ -796,9 +797,7 @@ fn main() -> Result<()> {
             }
             NeonLocalCmd::StorageBroker(subcmd) => rt.block_on(handle_storage_broker(&subcmd, env)),
             NeonLocalCmd::Safekeeper(subcmd) => rt.block_on(handle_safekeeper(&subcmd, env)),
-            NeonLocalCmd::EndpointStorage(subcmd) => {
-                rt.block_on(handle_endpoint_storage(&subcmd, env))
-            }
+            NeonLocalCmd::ObjectStorage(subcmd) => rt.block_on(handle_object_storage(&subcmd, env)),
             NeonLocalCmd::Endpoint(subcmd) => rt.block_on(handle_endpoint(&subcmd, env)),
             NeonLocalCmd::Mappings(subcmd) => handle_mappings(&subcmd, env),
         };
@@ -1015,8 +1014,8 @@ fn handle_init(args: &InitCmdArgs) -> anyhow::Result<LocalEnv> {
                     }
                 })
                 .collect(),
-            endpoint_storage: EndpointStorageConf {
-                port: ENDPOINT_STORAGE_DEFAULT_PORT,
+            object_storage: ObjectStorageConf {
+                port: OBJECT_STORAGE_DEFAULT_PORT,
             },
             pg_distrib_dir: None,
             neon_distrib_dir: None,
@@ -1736,15 +1735,12 @@ async fn handle_safekeeper(subcmd: &SafekeeperCmd, env: &local_env::LocalEnv) ->
     Ok(())
 }
 
-async fn handle_endpoint_storage(
-    subcmd: &EndpointStorageCmd,
-    env: &local_env::LocalEnv,
-) -> Result<()> {
-    use EndpointStorageCmd::*;
-    let storage = EndpointStorage::from_env(env);
+async fn handle_object_storage(subcmd: &ObjectStorageCmd, env: &local_env::LocalEnv) -> Result<()> {
+    use ObjectStorageCmd::*;
+    let storage = ObjectStorage::from_env(env);
 
     // In tests like test_forward_compatibility or test_graceful_cluster_restart
-    // old neon binaries (without endpoint_storage) are present
+    // old neon binaries (without object_storage) are present
     if !storage.bin.exists() {
         eprintln!(
             "{} binary not found. Ignore if this is a compatibility test",
@@ -1754,13 +1750,13 @@ async fn handle_endpoint_storage(
     }
 
     match subcmd {
-        Start(EndpointStorageStartCmd { start_timeout }) => {
+        Start(ObjectStorageStartCmd { start_timeout }) => {
             if let Err(e) = storage.start(start_timeout).await {
-                eprintln!("endpoint_storage start failed: {e}");
+                eprintln!("object_storage start failed: {e}");
                 exit(1);
             }
         }
-        Stop(EndpointStorageStopCmd { stop_mode }) => {
+        Stop(ObjectStorageStopCmd { stop_mode }) => {
             let immediate = match stop_mode {
                 StopMode::Fast => false,
                 StopMode::Immediate => true,
@@ -1870,10 +1866,10 @@ async fn handle_start_all_impl(
         }
 
         js.spawn(async move {
-            EndpointStorage::from_env(env)
+            ObjectStorage::from_env(env)
                 .start(&retry_timeout)
                 .await
-                .map_err(|e| e.context("start endpoint_storage"))
+                .map_err(|e| e.context("start object_storage"))
         });
     })();
 
@@ -1972,9 +1968,9 @@ async fn try_stop_all(env: &local_env::LocalEnv, immediate: bool) {
         }
     }
 
-    let storage = EndpointStorage::from_env(env);
+    let storage = ObjectStorage::from_env(env);
     if let Err(e) = storage.stop(immediate) {
-        eprintln!("endpoint_storage stop failed: {:#}", e);
+        eprintln!("object_storage stop failed: {:#}", e);
     }
 
     for ps_conf in &env.pageservers {

control_plane/src/lib.rs

@@ -9,8 +9,8 @@
 mod background_process;
 pub mod broker;
 pub mod endpoint;
-pub mod endpoint_storage;
 pub mod local_env;
+pub mod object_storage;
 pub mod pageserver;
 pub mod postgresql_conf;
 pub mod safekeeper;

control_plane/src/local_env.rs

@@ -19,7 +19,7 @@ use serde::{Deserialize, Serialize};
 use utils::auth::encode_from_key_file;
 use utils::id::{NodeId, TenantId, TenantTimelineId, TimelineId};
 
-use crate::endpoint_storage::{ENDPOINT_STORAGE_REMOTE_STORAGE_DIR, EndpointStorage};
+use crate::object_storage::{OBJECT_STORAGE_REMOTE_STORAGE_DIR, ObjectStorage};
 use crate::pageserver::{PAGESERVER_REMOTE_STORAGE_DIR, PageServerNode};
 use crate::safekeeper::SafekeeperNode;
 
@@ -72,7 +72,7 @@ pub struct LocalEnv {
 
     pub safekeepers: Vec<SafekeeperConf>,
 
-    pub endpoint_storage: EndpointStorageConf,
+    pub object_storage: ObjectStorageConf,
 
     // Control plane upcall API for pageserver: if None, we will not run storage_controller  If set, this will
     // be propagated into each pageserver's configuration.
@@ -110,7 +110,7 @@ pub struct OnDiskConfig {
     )]
     pub pageservers: Vec<PageServerConf>,
     pub safekeepers: Vec<SafekeeperConf>,
-    pub endpoint_storage: EndpointStorageConf,
+    pub object_storage: ObjectStorageConf,
     pub control_plane_api: Option<Url>,
     pub control_plane_hooks_api: Option<Url>,
     pub control_plane_compute_hook_api: Option<Url>,
@@ -144,7 +144,7 @@ pub struct NeonLocalInitConf {
     pub storage_controller: Option<NeonStorageControllerConf>,
     pub pageservers: Vec<NeonLocalInitPageserverConf>,
     pub safekeepers: Vec<SafekeeperConf>,
-    pub endpoint_storage: EndpointStorageConf,
+    pub object_storage: ObjectStorageConf,
     pub control_plane_api: Option<Url>,
     pub control_plane_hooks_api: Option<Url>,
     pub generate_local_ssl_certs: bool,
@@ -152,7 +152,7 @@ pub struct NeonLocalInitConf {
 
 #[derive(Serialize, Default, Deserialize, PartialEq, Eq, Clone, Debug)]
 #[serde(default)]
-pub struct EndpointStorageConf {
+pub struct ObjectStorageConf {
     pub port: u16,
 }
 
@@ -413,8 +413,8 @@ impl LocalEnv {
         self.pg_dir(pg_version, "lib")
     }
 
-    pub fn endpoint_storage_bin(&self) -> PathBuf {
-        self.neon_distrib_dir.join("endpoint_storage")
+    pub fn object_storage_bin(&self) -> PathBuf {
+        self.neon_distrib_dir.join("object_storage")
     }
 
     pub fn pageserver_bin(&self) -> PathBuf {
@@ -450,8 +450,8 @@ impl LocalEnv {
         self.base_data_dir.join("safekeepers").join(data_dir_name)
     }
 
-    pub fn endpoint_storage_data_dir(&self) -> PathBuf {
-        self.base_data_dir.join("endpoint_storage")
+    pub fn object_storage_data_dir(&self) -> PathBuf {
+        self.base_data_dir.join("object_storage")
     }
 
     pub fn get_pageserver_conf(&self, id: NodeId) -> anyhow::Result<&PageServerConf> {
@@ -615,7 +615,7 @@ impl LocalEnv {
                 control_plane_compute_hook_api: _,
                 branch_name_mappings,
                 generate_local_ssl_certs,
-                endpoint_storage,
+                object_storage,
             } = on_disk_config;
             LocalEnv {
                 base_data_dir: repopath.to_owned(),
@@ -632,7 +632,7 @@ impl LocalEnv {
                 control_plane_hooks_api,
                 branch_name_mappings,
                 generate_local_ssl_certs,
-                endpoint_storage,
+                object_storage,
             }
         };
 
@@ -742,7 +742,7 @@ impl LocalEnv {
                 control_plane_compute_hook_api: None,
                 branch_name_mappings: self.branch_name_mappings.clone(),
                 generate_local_ssl_certs: self.generate_local_ssl_certs,
-                endpoint_storage: self.endpoint_storage.clone(),
+                object_storage: self.object_storage.clone(),
             },
         )
     }
@@ -849,7 +849,7 @@ impl LocalEnv {
             control_plane_api,
             generate_local_ssl_certs,
             control_plane_hooks_api,
-            endpoint_storage,
+            object_storage,
         } = conf;
 
         // Find postgres binaries.
@@ -901,7 +901,7 @@ impl LocalEnv {
             control_plane_hooks_api,
             branch_name_mappings: Default::default(),
             generate_local_ssl_certs,
-            endpoint_storage,
+            object_storage,
         };
 
         if generate_local_ssl_certs {
@@ -929,13 +929,13 @@ impl LocalEnv {
                 .context("pageserver init failed")?;
         }
 
-        EndpointStorage::from_env(&env)
+        ObjectStorage::from_env(&env)
             .init()
             .context("object storage init failed")?;
 
         // setup remote remote location for default LocalFs remote storage
         std::fs::create_dir_all(env.base_data_dir.join(PAGESERVER_REMOTE_STORAGE_DIR))?;
-        std::fs::create_dir_all(env.base_data_dir.join(ENDPOINT_STORAGE_REMOTE_STORAGE_DIR))?;
+        std::fs::create_dir_all(env.base_data_dir.join(OBJECT_STORAGE_REMOTE_STORAGE_DIR))?;
 
         env.persist_config()
     }

control_plane/src/object_storage.rs

@@ -1,33 +1,34 @@
 use crate::background_process::{self, start_process, stop_process};
 use crate::local_env::LocalEnv;
+use anyhow::anyhow;
 use anyhow::{Context, Result};
 use camino::Utf8PathBuf;
 use std::io::Write;
 use std::time::Duration;
 
 /// Directory within .neon which will be used by default for LocalFs remote storage.
-pub const ENDPOINT_STORAGE_REMOTE_STORAGE_DIR: &str = "local_fs_remote_storage/endpoint_storage";
-pub const ENDPOINT_STORAGE_DEFAULT_PORT: u16 = 9993;
+pub const OBJECT_STORAGE_REMOTE_STORAGE_DIR: &str = "local_fs_remote_storage/object_storage";
+pub const OBJECT_STORAGE_DEFAULT_PORT: u16 = 9993;
 
-pub struct EndpointStorage {
+pub struct ObjectStorage {
     pub bin: Utf8PathBuf,
     pub data_dir: Utf8PathBuf,
     pub pemfile: Utf8PathBuf,
     pub port: u16,
 }
 
-impl EndpointStorage {
-    pub fn from_env(env: &LocalEnv) -> EndpointStorage {
-        EndpointStorage {
-            bin: Utf8PathBuf::from_path_buf(env.endpoint_storage_bin()).unwrap(),
-            data_dir: Utf8PathBuf::from_path_buf(env.endpoint_storage_data_dir()).unwrap(),
+impl ObjectStorage {
+    pub fn from_env(env: &LocalEnv) -> ObjectStorage {
+        ObjectStorage {
+            bin: Utf8PathBuf::from_path_buf(env.object_storage_bin()).unwrap(),
+            data_dir: Utf8PathBuf::from_path_buf(env.object_storage_data_dir()).unwrap(),
             pemfile: Utf8PathBuf::from_path_buf(env.public_key_path.clone()).unwrap(),
-            port: env.endpoint_storage.port,
+            port: env.object_storage.port,
         }
     }
 
     fn config_path(&self) -> Utf8PathBuf {
-        self.data_dir.join("endpoint_storage.json")
+        self.data_dir.join("object_storage.json")
     }
 
     fn listen_addr(&self) -> Utf8PathBuf {
@@ -48,7 +49,7 @@ impl EndpointStorage {
         let cfg = Cfg {
             listen: self.listen_addr(),
             pemfile: parent.join(self.pemfile.clone()),
-            local_path: parent.join(ENDPOINT_STORAGE_REMOTE_STORAGE_DIR),
+            local_path: parent.join(OBJECT_STORAGE_REMOTE_STORAGE_DIR),
             r#type: "LocalFs".to_string(),
         };
         std::fs::create_dir_all(self.config_path().parent().unwrap())?;
@@ -58,19 +59,24 @@ impl EndpointStorage {
     }
 
     pub async fn start(&self, retry_timeout: &Duration) -> Result<()> {
-        println!("Starting endpoint_storage at {}", self.listen_addr());
+        println!("Starting s3 proxy at {}", self.listen_addr());
         std::io::stdout().flush().context("flush stdout")?;
 
         let process_status_check = || async {
-            let res = reqwest::Client::new().get(format!("http://{}/metrics", self.listen_addr()));
-            match res.send().await {
-                Ok(res) => Ok(res.status().is_success()),
-                Err(_) => Ok(false),
+            tokio::time::sleep(Duration::from_millis(500)).await;
+            let res = reqwest::Client::new()
+                .get(format!("http://{}/metrics", self.listen_addr()))
+                .send()
+                .await;
+            match res {
+                Ok(response) if response.status().is_success() => Ok(true),
+                Ok(_) => Err(anyhow!("Failed to query /metrics")),
+                Err(e) => Err(anyhow!("Failed to check node status: {e}")),
             }
         };
 
         let res = start_process(
-            "endpoint_storage",
+            "object_storage",
             &self.data_dir.clone().into_std_path_buf(),
             &self.bin.clone().into_std_path_buf(),
             vec![self.config_path().to_string()],
@@ -88,14 +94,14 @@ impl EndpointStorage {
     }
 
     pub fn stop(&self, immediate: bool) -> anyhow::Result<()> {
-        stop_process(immediate, "endpoint_storage", &self.pid_file())
+        stop_process(immediate, "object_storage", &self.pid_file())
     }
 
     fn log_file(&self) -> Utf8PathBuf {
-        self.data_dir.join("endpoint_storage.log")
+        self.data_dir.join("object_storage.log")
     }
 
     fn pid_file(&self) -> Utf8PathBuf {
-        self.data_dir.join("endpoint_storage.pid")
+        self.data_dir.join("object_storage.pid")
     }
 }

docs/rfcs/2025-04-23-architecture-naming-conventions.md

@@ -1,53 +0,0 @@
-# Architecture Naming Scheme
-
-## Summary
-
-Neon computes are going to support multiple CPU architectures, including 64-bit
-x86 and ARM. Architecture naming schemes are fairly inconsistent, at least when
-it comes to these two architectures in particular. Sometimes 64-bit x86 is known
-as `amd64` and at other times `x86_64`. For 64-bit ARM, it's a similar story
-with `arm64` and `aarch64`.
-
-## Motivation
-
-Consistency when referring to these architectures across the platform in
-important not only so that everyone is on the same page, but also because we
-have architecture-dependent portions of the platform. One notable example is
-remote extensions. Remote extensions need to be compiled and packaged into
-compressed tarballs that are then downloaded by computes on demand. Downloading
-a tarball for the wrong architecture will cause `dlopen(3)` to fail when
-Postgres attempts to load the library. The compressed tarballs are located in an
-S3 bucket with object keys of the form
-`$BUILD_TAG/$ARCH/$PG_VERSION_NUM/${EXTENSION_NAME}.tar.zst` to mitigate the
-potential for failure. The build and deployment pipeline for remote extensions
-needs to be in lock step with the code in the compute that actually fetches
-remote extensions.
-
-## Impacted Components
-
-- Control Plane when persisting the target architecture in compute flags if
-  specified.
-- Austocaling when scheduling compute pods.
-- `compute_ctl` when downloading remote extensions.
-- Build and deployment pipeline for remote extensions.
-
-## Prior Art for CPU Architecture Names
-
-- [Rust](https://doc.rust-lang.org/std/env/consts/constant.ARCH.html)
-  - `x86_64`
-  - `aarch64`
-- [Go](https://pkg.go.dev/internal/goarch#pkg-constants)
-  - `amd64`
-  - `arm64`
-- Kubernetes
-  - Because Kubernetes is written in Go, it inherits the naming scheme.
-
-Going all in on either the Rust naming scheme or the Go naming scheme can save
-us a branch when getting the architecture, so we should pick one and force the
-other side to conform.
-
-## Decision
-
-The heart of our platform is Kubernetes. We will inherit the Go naming
-scheme just like Kubernetes. This makes things easy for the autoscaling team
-when they do pod scheduling on nodes.

libs/compute_api/src/spec.rs

@@ -242,22 +242,13 @@ impl RemoteExtSpec {
 
         match self.extension_data.get(real_ext_name) {
             Some(_ext_data) => {
-                // We have decided to use the Go naming convention due to Kubernetes.
-
-                let arch = match std::env::consts::ARCH {
-                    "x86_64" => "amd64",
-                    "aarch64" => "arm64",
-                    arch => arch,
-                };
-
                 // Construct the path to the extension archive
                 // BUILD_TAG/PG_MAJOR_VERSION/extensions/EXTENSION_NAME.tar.zst
                 //
                 // Keep it in sync with path generation in
                 // https://github.com/neondatabase/build-custom-extensions/tree/main
-                let archive_path_str = format!(
-                    "{build_tag}/{arch}/{pg_major_version}/extensions/{real_ext_name}.tar.zst"
-                );
+                let archive_path_str =
+                    format!("{build_tag}/{pg_major_version}/extensions/{real_ext_name}.tar.zst");
                 Ok((
                     real_ext_name.to_string(),
                     RemotePath::from_string(&archive_path_str)?,

libs/remote_storage/Cargo.toml

@@ -18,7 +18,8 @@ camino = { workspace = true, features = ["serde1"] }
 humantime-serde.workspace = true
 hyper = { workspace = true, features = ["client"] }
 futures.workspace = true
-reqwest.workspace = true
+reqwest = { workspace = true, features = ["multipart", "stream"] }
+chrono = { version = "0.4", default-features = false, features = ["clock"] }
 serde.workspace = true
 serde_json.workspace = true
 tokio = { workspace = true, features = ["sync", "fs", "io-util"] }
@@ -40,6 +41,10 @@ http-types.workspace = true
 http-body-util.workspace = true
 itertools.workspace = true
 sync_wrapper = { workspace = true, features = ["futures"] }
+gcp_auth = "0.12.3"
+url.workspace = true
+http.workspace = true
+uuid.workspace = true
 
 [dev-dependencies]
 camino-tempfile.workspace = true

libs/remote_storage/src/config.rs

@@ -41,6 +41,7 @@ impl RemoteStorageKind {
             RemoteStorageKind::LocalFs { .. } => None,
             RemoteStorageKind::AwsS3(config) => Some(&config.bucket_name),
             RemoteStorageKind::AzureContainer(config) => Some(&config.container_name),
+            RemoteStorageKind::GCS(config) => Some(&config.bucket_name),
         }
     }
 }
@@ -51,6 +52,7 @@ impl RemoteStorageConfig {
         match &self.storage {
             RemoteStorageKind::LocalFs { .. } => DEFAULT_REMOTE_STORAGE_LOCALFS_CONCURRENCY_LIMIT,
             RemoteStorageKind::AwsS3(c) => c.concurrency_limit.into(),
+            RemoteStorageKind::GCS(c) => c.concurrency_limit.into(),
             RemoteStorageKind::AzureContainer(c) => c.concurrency_limit.into(),
         }
     }
@@ -85,6 +87,9 @@ pub enum RemoteStorageKind {
     /// Azure Blob based storage, storing all files in the container
     /// specified by the config
     AzureContainer(AzureConfig),
+    /// Google Cloud based storage, storing all files in the GCS bucket
+    /// specified by the config
+    GCS(GCSConfig),
 }
 
 /// AWS S3 bucket coordinates and access credentials to manage the bucket contents (read and write).
@@ -154,6 +159,32 @@ impl Debug for S3Config {
     }
 }
 
+#[derive(Clone, PartialEq, Eq, Deserialize, Serialize)]
+pub struct GCSConfig {
+    /// Name of the bucket to connect to.
+    pub bucket_name: String,
+    /// A "subfolder" in the bucket, to use the same bucket separately by multiple remote storage users at once.
+    pub prefix_in_bucket: Option<String>,
+    #[serde(default = "default_remote_storage_s3_concurrency_limit")]
+    pub concurrency_limit: NonZeroUsize,
+    #[serde(default = "default_max_keys_per_list_response")]
+    pub max_keys_per_list_response: Option<i32>,
+}
+
+impl Debug for GCSConfig {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("GCSConfig")
+            .field("bucket_name", &self.bucket_name)
+            .field("prefix_in_bucket", &self.prefix_in_bucket)
+            .field("concurrency_limit", &self.concurrency_limit)
+            .field(
+                "max_keys_per_list_response",
+                &self.max_keys_per_list_response,
+            )
+            .finish()
+    }
+}
+
 /// Azure  bucket coordinates and access credentials to manage the bucket contents (read and write).
 #[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct AzureConfig {
@@ -268,6 +299,30 @@ timeout = '5s'";
         );
     }
 
+    #[test]
+    fn test_gcs_parsing() {
+        let toml = "\
+    bucket_name = 'foo-bar'
+    prefix_in_bucket = '/pageserver'
+    ";
+
+        let config = parse(toml).unwrap();
+
+        assert_eq!(
+            config,
+            RemoteStorageConfig {
+                storage: RemoteStorageKind::GCS(GCSConfig {
+                    bucket_name: "foo-bar".into(),
+                    prefix_in_bucket: Some("pageserver/".into()),
+                    max_keys_per_list_response: DEFAULT_MAX_KEYS_PER_LIST_RESPONSE,
+                    concurrency_limit: std::num::NonZero::new(100).unwrap(),
+                }),
+                timeout: Duration::from_secs(120),
+                small_timeout: RemoteStorageConfig::DEFAULT_SMALL_TIMEOUT
+            }
+        );
+    }
+
     #[test]
     fn test_s3_parsing() {
         let toml = "\

libs/remote_storage/src/gcs_bucket.rs

@@ -0,0 +1,978 @@
+#![allow(dead_code)]
+#![allow(unused)]
+
+use crate::config::GCSConfig;
+use crate::error::Cancelled;
+pub(super) use crate::metrics::RequestKind;
+use crate::metrics::{AttemptOutcome, start_counting_cancelled_wait, start_measuring_requests};
+use crate::{
+    ConcurrencyLimiter, Download, DownloadError, DownloadOpts, GCS_SCOPES, Listing, ListingMode,
+    ListingObject, MAX_KEYS_PER_DELETE_GCS, REMOTE_STORAGE_PREFIX_SEPARATOR, RemotePath,
+    RemoteStorage, StorageMetadata, TimeTravelError, TimeoutOrCancel,
+};
+use anyhow::Context;
+use azure_core::Etag;
+use bytes::Bytes;
+use bytes::BytesMut;
+use chrono::DateTime;
+use futures::stream::Stream;
+use futures::stream::TryStreamExt;
+use futures_util::StreamExt;
+use gcp_auth::{Token, TokenProvider};
+use http::Method;
+use http::StatusCode;
+use reqwest::{Client, header};
+use scopeguard::ScopeGuard;
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use std::fmt::Debug;
+use std::num::NonZeroU32;
+use std::pin::{Pin, pin};
+use std::sync::Arc;
+use std::time::Duration;
+use std::time::SystemTime;
+use tokio_util::codec::{BytesCodec, FramedRead};
+use tokio_util::sync::CancellationToken;
+use tracing;
+use url::Url;
+use uuid::Uuid;
+
+// ---------
+pub struct GCSBucket {
+    token_provider: Arc<dyn TokenProvider>,
+    bucket_name: String,
+    prefix_in_bucket: Option<String>,
+    max_keys_per_list_response: Option<i32>,
+    concurrency_limiter: ConcurrencyLimiter,
+    pub timeout: Duration,
+}
+
+struct GetObjectRequest {
+    bucket: String,
+    key: String,
+    etag: Option<String>,
+    range: Option<String>,
+}
+
+// ---------
+
+impl GCSBucket {
+    pub async fn new(remote_storage_config: &GCSConfig, timeout: Duration) -> anyhow::Result<Self> {
+        tracing::debug!(
+            "creating remote storage for gcs bucket {}",
+            remote_storage_config.bucket_name
+        );
+
+        // clean up 'prefix_in_bucket' if user provides '/pageserver' or 'pageserver/'
+        let prefix_in_bucket = remote_storage_config
+            .prefix_in_bucket
+            .as_deref()
+            .map(|prefix| {
+                let mut prefix = prefix;
+                while prefix.starts_with(REMOTE_STORAGE_PREFIX_SEPARATOR) {
+                    prefix = &prefix[1..];
+                }
+
+                let mut prefix = prefix.to_string();
+                if prefix.ends_with(REMOTE_STORAGE_PREFIX_SEPARATOR) {
+                    prefix.pop();
+                }
+
+                prefix
+            });
+
+        // get GOOGLE_APPLICATION_CREDENTIALS
+        let provider = gcp_auth::provider().await?;
+
+        Ok(GCSBucket {
+            token_provider: Arc::clone(&provider),
+            bucket_name: remote_storage_config.bucket_name.clone(),
+            prefix_in_bucket,
+            timeout,
+            max_keys_per_list_response: remote_storage_config.max_keys_per_list_response,
+            concurrency_limiter: ConcurrencyLimiter::new(
+                remote_storage_config.concurrency_limit.get(),
+            ),
+        })
+    }
+
+    // convert `RemotePath` -> `String`
+    pub fn relative_path_to_gcs_object(&self, path: &RemotePath) -> String {
+        let path_string = path.get_path().as_str();
+        match &self.prefix_in_bucket {
+            Some(prefix) => prefix.clone() + "/" + path_string,
+            None => path_string.to_string(),
+        }
+    }
+
+    // convert `String` -> `RemotePath`
+    pub fn gcs_object_to_relative_path(&self, key: &str) -> RemotePath {
+        let relative_path =
+            match key.strip_prefix(self.prefix_in_bucket.as_deref().unwrap_or_default()) {
+                Some(stripped) => stripped,
+                // we rely on GCS to return properly prefixed paths
+                // for requests with a certain prefix
+                None => panic!(
+                    "Key {} does not start with bucket prefix {:?}",
+                    key, self.prefix_in_bucket
+                ),
+            };
+        RemotePath(
+            relative_path
+                .split(REMOTE_STORAGE_PREFIX_SEPARATOR)
+                .collect(),
+        )
+    }
+
+    pub fn bucket_name(&self) -> &str {
+        &self.bucket_name
+    }
+
+    fn max_keys_per_delete(&self) -> usize {
+        MAX_KEYS_PER_DELETE_GCS
+    }
+
+    async fn permit(
+        &self,
+        kind: RequestKind,
+        cancel: &CancellationToken,
+    ) -> Result<tokio::sync::SemaphorePermit<'_>, Cancelled> {
+        let started_at = start_counting_cancelled_wait(kind);
+        let acquire = self.concurrency_limiter.acquire(kind);
+
+        let permit = tokio::select! {
+            permit = acquire => permit.expect("semaphore is never closed"),
+            _ = cancel.cancelled() => return Err(Cancelled),
+        };
+
+        let started_at = ScopeGuard::into_inner(started_at);
+        crate::metrics::BUCKET_METRICS
+            .wait_seconds
+            .observe_elapsed(kind, started_at);
+
+        Ok(permit)
+    }
+
+    async fn owned_permit(
+        &self,
+        kind: RequestKind,
+        cancel: &CancellationToken,
+    ) -> Result<tokio::sync::OwnedSemaphorePermit, Cancelled> {
+        let started_at = start_counting_cancelled_wait(kind);
+        let acquire = self.concurrency_limiter.acquire_owned(kind);
+
+        let permit = tokio::select! {
+            permit = acquire => permit.expect("semaphore is never closed"),
+            _ = cancel.cancelled() => return Err(Cancelled),
+        };
+
+        let started_at = ScopeGuard::into_inner(started_at);
+        crate::metrics::BUCKET_METRICS
+            .wait_seconds
+            .observe_elapsed(kind, started_at);
+        Ok(permit)
+    }
+
+    async fn put_object(
+        &self,
+        byte_stream: impl Stream<Item = std::io::Result<Bytes>> + Send + Sync + 'static,
+        fs_size: usize,
+        to: &RemotePath,
+        cancel: &CancellationToken,
+    ) -> anyhow::Result<()> {
+        // https://cloud.google.com/storage/docs/xml-api/reference-headers#chunked
+        let mut headers = header::HeaderMap::new();
+        headers.insert(
+            header::TRANSFER_ENCODING,
+            header::HeaderValue::from_static("chunked"),
+        );
+
+        // TODO Check if we need type 'multipart/related' file to attach metadata like Neon's S3
+        // `.upload()` does.
+        // https://cloud.google.com/storage/docs/uploading-objects#uploading-an-object
+        let upload_uri = format!(
+            "https://storage.googleapis.com/upload/storage/v1/b/{}/o/?uploadType=media&name={}",
+            self.bucket_name.clone(),
+            self.relative_path_to_gcs_object(to).trim_start_matches("/")
+        );
+
+        let upload = Client::new()
+            .post(upload_uri)
+            .body(reqwest::Body::wrap_stream(byte_stream))
+            .headers(headers)
+            .bearer_auth(self.token_provider.token(GCS_SCOPES).await?.as_str())
+            .send();
+
+        // We await it in a race against the Tokio timeout
+        let upload = tokio::time::timeout(self.timeout, upload);
+        let res = tokio::select! {
+            res = upload => res,
+            _ = cancel.cancelled() => return Err(TimeoutOrCancel::Cancel.into()),
+        };
+
+        match res {
+            Ok(Ok(res)) => {
+                if !res.status().is_success() {
+                    match res.status() {
+                        StatusCode::NOT_FOUND => {
+                            return Err(anyhow::anyhow!("GCS error: not found \n\t {:?}", res));
+                        }
+                        _ => {
+                            return Err(anyhow::anyhow!(
+                                "GCS PUT response contained no response body \n\t {:?}",
+                                res
+                            ));
+                        }
+                    }
+                } else {
+                    Ok(())
+                }
+            }
+            Ok(Err(reqw)) => Err(reqw.into()),
+            Err(_timeout) => Err(TimeoutOrCancel::Timeout.into()),
+        }
+    }
+
+    async fn copy(
+        &self,
+        from: String,
+        to: String,
+        cancel: &CancellationToken,
+    ) -> anyhow::Result<()> {
+        let kind = RequestKind::Copy;
+
+        let _permit = self.permit(kind, cancel).await?;
+
+        let timeout = tokio::time::sleep(self.timeout);
+
+        let started_at = start_measuring_requests(kind);
+
+        let copy_uri = format!(
+            "https://storage.googleapis.com/storage/v1/b/{}/o/{}/copyTo/b/{}/o/{}",
+            self.bucket_name.clone(),
+            &from,
+            self.bucket_name.clone(),
+            &to
+        );
+
+        let op = Client::new()
+            .post(copy_uri)
+            .bearer_auth(self.token_provider.token(GCS_SCOPES).await?.as_str())
+            .send();
+
+        let res = tokio::select! {
+            res = op => res,
+            _ = timeout => return Err(TimeoutOrCancel::Timeout.into()),
+            _ = cancel.cancelled() => return Err(TimeoutOrCancel::Cancel.into()),
+        };
+
+        let started_at = ScopeGuard::into_inner(started_at);
+        crate::metrics::BUCKET_METRICS
+            .req_seconds
+            .observe_elapsed(kind, &res, started_at);
+
+        res?;
+
+        Ok(())
+    }
+
+    async fn delete_oids(
+        &self,
+        delete_objects: &[String],
+        cancel: &CancellationToken,
+        _permit: &tokio::sync::SemaphorePermit<'_>,
+    ) -> anyhow::Result<()> {
+        let kind = RequestKind::Delete;
+        let mut cancel = std::pin::pin!(cancel.cancelled());
+
+        for chunk in delete_objects.chunks(MAX_KEYS_PER_DELETE_GCS) {
+            let started_at = start_measuring_requests(kind);
+
+            // Use this to report keys that didn't delete based on 'content_id'
+            let mut delete_objects_status = HashMap::new();
+
+            let mut form = reqwest::multipart::Form::new();
+            let bulk_uri = "https://storage.googleapis.com/batch/storage/v1";
+
+            for (index, path) in delete_objects.iter().enumerate() {
+                delete_objects_status.insert(index + 1, path.clone());
+
+                let path_to_delete: String =
+                    url::form_urlencoded::byte_serialize(path.trim_start_matches("/").as_bytes())
+                        .collect();
+
+                let delete_req = format!(
+                    "
+                    DELETE /storage/v1/b/{}/o/{} HTTP/1.1\r\n\
+                    Content-Type: application/json\r\n\
+                    accept: application/json\r\n\
+                    content-length: 0\r\n
+                    ",
+                    self.bucket_name.clone(),
+                    path_to_delete
+                )
+                .trim()
+                .to_string();
+
+                let content_id = format!("<{}+{}>", Uuid::new_v4(), index + 1);
+
+                let mut part_headers = header::HeaderMap::new();
+                part_headers.insert(
+                    header::CONTENT_TYPE,
+                    header::HeaderValue::from_static("application/http"),
+                );
+                part_headers.insert(
+                    header::TRANSFER_ENCODING,
+                    header::HeaderValue::from_static("binary"),
+                );
+                part_headers.insert(
+                    header::HeaderName::from_static("content-id"),
+                    header::HeaderValue::from_str(&content_id)?,
+                );
+                let part = reqwest::multipart::Part::text(delete_req).headers(part_headers);
+
+                form = form.part(format!("request-{}", index), part);
+            }
+
+            let mut headers = header::HeaderMap::new();
+            headers.insert(
+                header::CONTENT_TYPE,
+                header::HeaderValue::from_str(&format!(
+                    "multipart/mixed; boundary={}",
+                    form.boundary()
+                ))?,
+            );
+
+            let req = Client::new()
+                .post(bulk_uri)
+                .bearer_auth(self.token_provider.token(GCS_SCOPES).await?.as_str())
+                .multipart(form)
+                .headers(headers)
+                .send();
+
+            let resp = tokio::select! {
+                resp = req => resp,
+                _ = tokio::time::sleep(self.timeout) => return Err(TimeoutOrCancel::Timeout.into()),
+                _ = &mut cancel => return Err(TimeoutOrCancel::Cancel.into()),
+            };
+
+            let started_at = ScopeGuard::into_inner(started_at);
+            crate::metrics::BUCKET_METRICS
+                .req_seconds
+                .observe_elapsed(kind, &resp, started_at);
+
+            let resp = resp.context("request deletion")?;
+
+            crate::metrics::BUCKET_METRICS
+                .deleted_objects_total
+                .inc_by(chunk.len() as u64);
+
+            let res_headers = resp.headers().to_owned();
+
+            let boundary = res_headers
+                .get(header::CONTENT_TYPE)
+                .unwrap()
+                .to_str()?
+                .split("=")
+                .last()
+                .unwrap();
+
+            let res_body = resp.text().await?;
+
+            let parsed: HashMap<String, String> = res_body
+                .split(&format!("--{}", boundary))
+                .filter_map(|c| {
+                    let mut lines = c.lines();
+
+                    let id = lines.find_map(|line| {
+                        line.strip_prefix("Content-ID:")
+                            .and_then(|suf| suf.split('+').last())
+                            .and_then(|suf| suf.split('>').next())
+                            .map(|x| x.trim().to_string())
+                    });
+
+                    let status_code = lines.find_map(|line| {
+                        // Not sure if this protocol version shouldn't be so specific
+                        line.strip_prefix("HTTP/1.1")
+                            .and_then(|x| x.split_whitespace().next())
+                            .map(|x| x.trim().to_string())
+                    });
+
+                    id.zip(status_code)
+                })
+                .collect();
+
+            // Gather failures
+            let errors: HashMap<usize, &String> = parsed
+                .iter()
+                .filter_map(|(x, y)| {
+                    if y.chars().next() != Some('2') {
+                        x.parse::<usize>().ok().map(|v| (v, y))
+                    } else {
+                        None
+                    }
+                })
+                .collect();
+
+            if !errors.is_empty() {
+                // Report 10 of them like S3
+                const LOG_UP_TO_N_ERRORS: usize = 10;
+                for (id, code) in errors.iter().take(LOG_UP_TO_N_ERRORS) {
+                    tracing::warn!(
+                        "DeleteObjects key {} failed with code: {}",
+                        delete_objects_status.get(id).unwrap(),
+                        code
+                    );
+                }
+
+                return Err(anyhow::anyhow!(
+                    "Failed to delete {}/{} objects",
+                    errors.len(),
+                    chunk.len(),
+                ));
+            }
+        }
+
+        Ok(())
+    }
+
+    async fn list_objects_v2(&self, list_uri: String) -> anyhow::Result<reqwest::RequestBuilder> {
+        let res = Client::new()
+            .get(list_uri)
+            .bearer_auth(self.token_provider.token(GCS_SCOPES).await?.as_str());
+        Ok(res)
+    }
+
+    // need a 'bucket', a 'key', and a bytes 'range'.
+    async fn get_object(
+        &self,
+        request: GetObjectRequest,
+        cancel: &CancellationToken,
+    ) -> anyhow::Result<Download, DownloadError> {
+        let kind = RequestKind::Get;
+
+        let permit = self.owned_permit(kind, cancel).await?;
+
+        let started_at = start_measuring_requests(kind);
+
+        let encoded_path: String =
+            url::form_urlencoded::byte_serialize(request.key.as_bytes()).collect();
+
+        /// We do this in two parts:
+        /// 1. Serialize the metadata of the first request to get Etag, last modified, etc
+        /// 2. We do not .await the second request pass on the pinned stream to the 'get_object'
+        ///    caller
+        // 1. Serialize Metadata in initial request
+        let metadata_uri_mod = "alt=json";
+        let download_uri = format!(
+            "https://storage.googleapis.com/storage/v1/b/{}/o/{}?{}",
+            self.bucket_name.clone(),
+            encoded_path,
+            metadata_uri_mod
+        );
+
+        let res = Client::new()
+            .get(download_uri)
+            .bearer_auth(
+                self.token_provider
+                    .token(GCS_SCOPES)
+                    .await
+                    .map_err(|e: gcp_auth::Error| DownloadError::Other(e.into()))?
+                    .as_str(),
+            )
+            .send()
+            .await
+            .map_err(|e: reqwest::Error| DownloadError::Other(e.into()))?;
+
+        if !res.status().is_success() {
+            match res.status() {
+                StatusCode::NOT_FOUND => return Err(DownloadError::NotFound),
+                _ => {
+                    return Err(DownloadError::Other(anyhow::anyhow!(
+                        "GCS GET resposne contained no response body"
+                    )));
+                }
+            }
+        };
+
+        let body = res
+            .text()
+            .await
+            .map_err(|e: reqwest::Error| DownloadError::Other(e.into()))?;
+
+        let resp: GCSObject = serde_json::from_str(&body)
+            .map_err(|e: serde_json::Error| DownloadError::Other(e.into()))?;
+
+        // 2. Byte Stream request
+        let mut headers = header::HeaderMap::new();
+        headers.insert(header::RANGE, header::HeaderValue::from_static("bytes=0-"));
+
+        let encoded_path: String =
+            url::form_urlencoded::byte_serialize(request.key.as_bytes()).collect();
+
+        let stream_uri_mod = "alt=media";
+        let stream_uri = format!(
+            "https://storage.googleapis.com/storage/v1/b/{}/o/{}?{}",
+            self.bucket_name.clone(),
+            encoded_path,
+            stream_uri_mod
+        );
+
+        let mut req = Client::new()
+            .get(stream_uri)
+            .headers(headers)
+            .bearer_auth(
+                self.token_provider
+                    .token(GCS_SCOPES)
+                    .await
+                    .map_err(|e: gcp_auth::Error| DownloadError::Other(e.into()))?
+                    .as_str(),
+            )
+            .send();
+
+        let get_object = tokio::select! {
+            res = req => res,
+            _ = tokio::time::sleep(self.timeout) => return Err(DownloadError::Timeout),
+            _ = cancel.cancelled() => return Err(DownloadError::Cancelled),
+        };
+
+        let started_at = ScopeGuard::into_inner(started_at);
+
+        let object_output = match get_object {
+            Ok(object_output) => {
+                if !object_output.status().is_success() {
+                    match object_output.status() {
+                        StatusCode::NOT_FOUND => return Err(DownloadError::NotFound),
+                        _ => {
+                            return Err(DownloadError::Other(anyhow::anyhow!(
+                                "GCS GET resposne contained no response body"
+                            )));
+                        }
+                    }
+                } else {
+                    object_output
+                }
+            }
+            Err(e) => {
+                crate::metrics::BUCKET_METRICS.req_seconds.observe_elapsed(
+                    kind,
+                    AttemptOutcome::Err,
+                    started_at,
+                );
+
+                return Err(DownloadError::Other(
+                    anyhow::Error::new(e).context("download s3 object"),
+                ));
+            }
+        };
+
+        let remaining = self.timeout.saturating_sub(started_at.elapsed());
+
+        let metadata = resp.metadata.map(StorageMetadata);
+
+        let etag = resp
+            .etag
+            .ok_or(DownloadError::Other(anyhow::anyhow!("Missing ETag header")))?
+            .into();
+
+        let last_modified: SystemTime = resp
+            .updated
+            .and_then(|s| DateTime::parse_from_rfc3339(&s).ok())
+            .map(|s| s.into())
+            .unwrap_or(SystemTime::now());
+
+        // But let data stream pass through
+        Ok(Download {
+            download_stream: Box::pin(object_output.bytes_stream().map(|item| {
+                item.map_err(|e: reqwest::Error| std::io::Error::new(std::io::ErrorKind::Other, e))
+            })),
+            etag,
+            last_modified,
+            metadata,
+        })
+    }
+}
+
+impl RemoteStorage for GCSBucket {
+    // ---------------------------------------
+    // Neon wrappers for GCS client functions
+    // ---------------------------------------
+
+    fn list_streaming(
+        &self,
+        prefix: Option<&RemotePath>,
+        mode: ListingMode,
+        max_keys: Option<NonZeroU32>,
+        cancel: &CancellationToken,
+    ) -> impl Stream<Item = Result<Listing, DownloadError>> {
+        let kind = RequestKind::List;
+
+        let mut max_keys = max_keys.map(|mk| mk.get() as i32);
+
+        let list_prefix = prefix
+            .map(|p| self.relative_path_to_gcs_object(p))
+            .or_else(|| {
+                self.prefix_in_bucket.clone().map(|mut s| {
+                    s.push(REMOTE_STORAGE_PREFIX_SEPARATOR);
+                    s
+                })
+            })
+            .unwrap();
+
+        let request_max_keys = self
+            .max_keys_per_list_response
+            .into_iter()
+            .chain(max_keys.into_iter())
+            .min()
+            // https://cloud.google.com/storage/docs/json_api/v1/objects/list?hl=en#parameters
+            // TODO set this to default
+            .unwrap_or(1000);
+
+        // We pass URI in to `list_objects_v2` as we'll modify it with `NextPageToken`, hence
+        // `mut`
+        let mut list_uri = format!(
+            "https://storage.googleapis.com/storage/v1/b/{}/o?prefix={}&maxResults={}",
+            self.bucket_name.clone(),
+            list_prefix,
+            request_max_keys,
+        );
+
+        // on ListingMode:
+        // https://github.com/neondatabase/neon/blob/edc11253b65e12a10843711bd88ad277511396d7/libs/remote_storage/src/lib.rs#L158C1-L164C2
+        if let ListingMode::WithDelimiter = mode {
+            list_uri.push_str(&format!(
+                "&delimiter={}",
+                REMOTE_STORAGE_PREFIX_SEPARATOR.to_string()
+            ));
+        }
+
+        async_stream::stream! {
+
+            let mut continuation_token = None;
+
+            'outer: loop {
+                let started_at = start_measuring_requests(kind);
+
+                let request = self.list_objects_v2(list_uri.clone())
+                    .await
+                    .map_err(DownloadError::Other)?
+                    .send();
+
+                // this is like `await`
+                let response = tokio::select! {
+                    res = request => Ok(res),
+                    _ = tokio::time::sleep(self.timeout) => Err(DownloadError::Timeout),
+                    _ = cancel.cancelled() => Err(DownloadError::Cancelled),
+                }?;
+
+                // just mapping our `Result' error variant's type.
+                let response = response
+                    .context("Failed to list GCS prefixes")
+                    .map_err(DownloadError::Other);
+
+                let started_at = ScopeGuard::into_inner(started_at);
+
+                crate::metrics::BUCKET_METRICS
+                    .req_seconds
+                    .observe_elapsed(kind, &response, started_at);
+
+                let response = match response {
+                    Ok(response) => response,
+                    Err(e) => {
+                        // The error is potentially retryable, so we must rewind the loop after yielding.
+                        yield Err(e);
+                        continue 'outer;
+                    },
+                };
+
+                let body = response.text()
+                    .await
+                    .map_err(|e: reqwest::Error| DownloadError::Other(e.into()))?;
+
+                let resp: GCSListResponse = serde_json::from_str(&body).map_err(|e: serde_json::Error| DownloadError::Other(e.into()))?;
+
+                let prefixes = resp.common_prefixes();
+                let keys = resp.contents();
+
+                tracing::debug!("list: {} prefixes, {} keys", prefixes.len(), keys.len());
+
+                let mut result = Listing::default();
+
+                for res in keys.iter() {
+
+                   let last_modified: SystemTime = res.updated.clone()
+                       .and_then(|s| DateTime::parse_from_rfc3339(&s).ok())
+                       .map(|s| s.into())
+                       .unwrap_or(SystemTime::now());
+
+                   let size = res.size.clone().unwrap_or("0".to_string()).parse::<u64>().unwrap();
+
+                   let key = res.name.clone();
+
+                   result.keys.push(
+                        ListingObject{
+                            key: self.gcs_object_to_relative_path(&key),
+                            last_modified,
+                            size,
+                        }
+                   );
+
+                   if let Some(mut mk) = max_keys {
+                       assert!(mk > 0);
+                       mk -= 1;
+                       if mk == 0 {
+                          tracing::debug!("reached limit set by max_keys");
+                          yield Ok(result);
+                          break 'outer;
+                       }
+                       max_keys = Some(mk);
+                   };
+                }
+
+                result.prefixes.extend(prefixes.iter().filter_map(|p| {
+                    Some(
+                        self.gcs_object_to_relative_path(
+                            p.trim_end_matches(REMOTE_STORAGE_PREFIX_SEPARATOR)
+                        ),
+                    )
+                }));
+
+                yield Ok(result);
+
+                continuation_token = match resp.next_page_token {
+                    Some(token) => {
+                        list_uri = list_uri + "&pageToken=" + &token;
+                        Some(token)
+                    },
+                    None => break
+                }
+            }
+        }
+    }
+
+    async fn head_object(
+        &self,
+        key: &RemotePath,
+        cancel: &CancellationToken,
+    ) -> Result<ListingObject, DownloadError> {
+        let kind = RequestKind::Head;
+
+        todo!();
+    }
+
+    async fn upload(
+        &self,
+        from: impl Stream<Item = std::io::Result<Bytes>> + Send + Sync + 'static,
+        from_size_bytes: usize,
+        to: &RemotePath,
+        metadata: Option<StorageMetadata>,
+        cancel: &CancellationToken,
+    ) -> anyhow::Result<()> {
+        let kind = RequestKind::Put;
+        let _permit = self.permit(kind, cancel).await?;
+
+        let started_at = start_measuring_requests(kind);
+
+        let upload = self.put_object(from, from_size_bytes, to, cancel);
+
+        let upload = tokio::time::timeout(self.timeout, upload);
+
+        let res = tokio::select! {
+            res = upload => res,
+            _ = cancel.cancelled() => return Err(TimeoutOrCancel::Cancel.into()),
+        };
+
+        if let Ok(inner) = &res {
+            // do not incl. timeouts as errors in metrics but cancellations
+            let started_at = ScopeGuard::into_inner(started_at);
+            crate::metrics::BUCKET_METRICS
+                .req_seconds
+                .observe_elapsed(kind, inner, started_at);
+        }
+
+        match res {
+            Ok(Ok(_put)) => Ok(()),
+            Ok(Err(sdk)) => Err(sdk.into()),
+            Err(_timeout) => Err(TimeoutOrCancel::Timeout.into()),
+        }
+    }
+
+    async fn copy(
+        &self,
+        from: &RemotePath,
+        to: &RemotePath,
+        cancel: &CancellationToken,
+    ) -> anyhow::Result<()> {
+        let kind = RequestKind::Copy;
+        let _permit = self.permit(kind, cancel).await?;
+
+        let timeout = tokio::time::sleep(self.timeout);
+
+        let started_at = start_measuring_requests(kind);
+
+        // we need to specify bucket_name as a prefix
+        let copy_source = format!(
+            "{}/{}",
+            self.bucket_name,
+            self.relative_path_to_gcs_object(from)
+        );
+
+        todo!();
+    }
+
+    async fn download(
+        &self,
+        from: &RemotePath,
+        opts: &DownloadOpts,
+        cancel: &CancellationToken,
+    ) -> Result<Download, DownloadError> {
+        // if prefix is not none then download file `prefix/from`
+        // if prefix is none then download file `from`
+
+        self.get_object(
+            GetObjectRequest {
+                bucket: self.bucket_name.clone(),
+                key: self
+                    .relative_path_to_gcs_object(from)
+                    .trim_start_matches("/")
+                    .to_string(),
+                etag: opts.etag.as_ref().map(|e| e.to_string()),
+                range: opts.byte_range_header(),
+            },
+            cancel,
+        )
+        .await
+    }
+
+    async fn delete_objects(
+        &self,
+        paths: &[RemotePath],
+        cancel: &CancellationToken,
+    ) -> anyhow::Result<()> {
+        let kind = RequestKind::Delete;
+        let permit = self.permit(kind, cancel).await?;
+
+        let mut delete_objects: Vec<String> = Vec::with_capacity(paths.len());
+
+        let delete_objects: Vec<String> = paths
+            .iter()
+            .map(|i| self.relative_path_to_gcs_object(i))
+            .collect();
+
+        self.delete_oids(&delete_objects, cancel, &permit).await
+    }
+
+    fn max_keys_per_delete(&self) -> usize {
+        MAX_KEYS_PER_DELETE_GCS
+    }
+
+    async fn delete(&self, path: &RemotePath, cancel: &CancellationToken) -> anyhow::Result<()> {
+        let paths = std::array::from_ref(path);
+        self.delete_objects(paths, cancel).await
+    }
+
+    async fn time_travel_recover(
+        &self,
+        prefix: Option<&RemotePath>,
+        timestamp: SystemTime,
+        done_if_after: SystemTime,
+        cancel: &CancellationToken,
+    ) -> Result<(), TimeTravelError> {
+        Ok(())
+    }
+}
+
+// ---------
+
+#[derive(Serialize, Deserialize, Debug)]
+#[serde(rename_all = "snake_case")]
+pub struct GCSListResponse {
+    #[serde(rename = "nextPageToken")]
+    pub next_page_token: Option<String>,
+    pub items: Option<Vec<GCSObject>>,
+    pub prefixes: Option<Vec<String>>,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+#[serde(rename_all = "snake_case")]
+pub struct GCSObject {
+    pub name: String,
+    pub bucket: String,
+    pub generation: String,
+    pub metageneration: String,
+    #[serde(rename = "contentType")]
+    pub content_type: Option<String>,
+    #[serde(rename = "storageClass")]
+    pub storage_class: String,
+    pub size: Option<String>,
+    #[serde(rename = "md5Hash")]
+    pub md5_hash: Option<String>,
+    pub crc32c: String,
+    pub etag: Option<String>,
+    #[serde(rename = "timeCreated")]
+    pub time_created: String,
+    pub updated: Option<String>,
+    #[serde(rename = "timeStorageClassUpdated")]
+    pub time_storage_class_updated: String,
+    #[serde(rename = "timeFinalized")]
+    pub time_finalized: String,
+    pub metadata: Option<HashMap<String, String>>,
+}
+
+impl GCSListResponse {
+    pub fn contents(&self) -> &[GCSObject] {
+        self.items.as_deref().unwrap_or_default()
+    }
+    pub fn common_prefixes(&self) -> &[String] {
+        self.prefixes.as_deref().unwrap_or_default()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+
+    use super::*;
+    use gcp_auth;
+    use std::num::NonZero;
+    use std::pin::pin;
+    use std::sync::Arc;
+
+    const BUFFER_SIZE: usize = 32 * 1024;
+
+    // TODO what does Neon want here for integration tests?
+    const BUCKET: &str = "https://storage.googleapis.com/storage/v1/b/my-test-bucket";
+
+    #[tokio::test]
+    async fn list_returns_keys_from_bucket() {
+        let provider = gcp_auth::provider().await.unwrap();
+        let gcs = GCSBucket {
+            token_provider: Arc::clone(&provider),
+            bucket_name: BUCKET.to_string(),
+            prefix_in_bucket: None,
+            max_keys_per_list_response: Some(100),
+            concurrency_limiter: ConcurrencyLimiter::new(100),
+            timeout: std::time::Duration::from_secs(120),
+        };
+
+        let cancel = CancellationToken::new();
+        let remote_prefix = "box/tiff/2023/TN".to_string();
+        let max_keys: u32 = 100;
+        let mut stream = pin!(gcs.list_streaming(Some(remote_prefix), NonZero::new(max_keys)));
+        let mut combined = stream
+            .next()
+            .await
+            .expect("At least one item required")
+            .unwrap();
+        while let Some(list) = stream.next().await {
+            let list = list.unwrap();
+            combined.keys.extend(list.keys.into_iter());
+            combined.prefixes.extend_from_slice(&list.prefixes);
+        }
+
+        for key in combined.keys.iter() {
+            println!("Item: {} -- {:?}", key.key, key.last_modified);
+        }
+
+        assert_ne!(0, combined.keys.len());
+    }
+}

libs/remote_storage/src/lib.rs

@@ -12,6 +12,7 @@
 mod azure_blob;
 mod config;
 mod error;
+mod gcs_bucket;
 mod local_fs;
 mod metrics;
 mod s3_bucket;
@@ -42,6 +43,7 @@ use tokio_util::sync::CancellationToken;
 use tracing::info;
 
 pub use self::azure_blob::AzureBlobStorage;
+pub use self::gcs_bucket::GCSBucket;
 pub use self::local_fs::LocalFs;
 pub use self::s3_bucket::S3Bucket;
 pub use self::simulate_failures::UnreliableWrapper;
@@ -80,8 +82,12 @@ pub const MAX_KEYS_PER_DELETE_S3: usize = 1000;
 /// <https://learn.microsoft.com/en-us/rest/api/storageservices/blob-batch>
 pub const MAX_KEYS_PER_DELETE_AZURE: usize = 256;
 
+pub const MAX_KEYS_PER_DELETE_GCS: usize = 1000;
+
 const REMOTE_STORAGE_PREFIX_SEPARATOR: char = '/';
 
+const GCS_SCOPES: &[&str] = &["https://www.googleapis.com/auth/cloud-platform"];
+
 /// Path on the remote storage, relative to some inner prefix.
 /// The prefix is an implementation detail, that allows representing local paths
 /// as the remote ones, stripping the local storage prefix away.
@@ -439,6 +445,7 @@ pub enum GenericRemoteStorage<Other: Clone = Arc<UnreliableWrapper>> {
     AwsS3(Arc<S3Bucket>),
     AzureBlob(Arc<AzureBlobStorage>),
     Unreliable(Other),
+    GCS(Arc<GCSBucket>),
 }
 
 impl<Other: RemoteStorage> GenericRemoteStorage<Arc<Other>> {
@@ -455,6 +462,7 @@ impl<Other: RemoteStorage> GenericRemoteStorage<Arc<Other>> {
             Self::AwsS3(s) => s.list(prefix, mode, max_keys, cancel).await,
             Self::AzureBlob(s) => s.list(prefix, mode, max_keys, cancel).await,
             Self::Unreliable(s) => s.list(prefix, mode, max_keys, cancel).await,
+            Self::GCS(s) => s.list(prefix, mode, max_keys, cancel).await,
         }
     }
 
@@ -472,6 +480,7 @@ impl<Other: RemoteStorage> GenericRemoteStorage<Arc<Other>> {
             Self::AwsS3(s) => Box::pin(s.list_streaming(prefix, mode, max_keys, cancel)),
             Self::AzureBlob(s) => Box::pin(s.list_streaming(prefix, mode, max_keys, cancel)),
             Self::Unreliable(s) => Box::pin(s.list_streaming(prefix, mode, max_keys, cancel)),
+            Self::GCS(s) => Box::pin(s.list_streaming(prefix, mode, max_keys, cancel)),
         }
     }
 
@@ -486,6 +495,7 @@ impl<Other: RemoteStorage> GenericRemoteStorage<Arc<Other>> {
             Self::AwsS3(s) => s.head_object(key, cancel).await,
             Self::AzureBlob(s) => s.head_object(key, cancel).await,
             Self::Unreliable(s) => s.head_object(key, cancel).await,
+            Self::GCS(_) => todo!(),
         }
     }
 
@@ -503,6 +513,7 @@ impl<Other: RemoteStorage> GenericRemoteStorage<Arc<Other>> {
             Self::AwsS3(s) => s.upload(from, data_size_bytes, to, metadata, cancel).await,
             Self::AzureBlob(s) => s.upload(from, data_size_bytes, to, metadata, cancel).await,
             Self::Unreliable(s) => s.upload(from, data_size_bytes, to, metadata, cancel).await,
+            Self::GCS(s) => s.upload(from, data_size_bytes, to, metadata, cancel).await,
         }
     }
 
@@ -518,6 +529,7 @@ impl<Other: RemoteStorage> GenericRemoteStorage<Arc<Other>> {
             Self::AwsS3(s) => s.download(from, opts, cancel).await,
             Self::AzureBlob(s) => s.download(from, opts, cancel).await,
             Self::Unreliable(s) => s.download(from, opts, cancel).await,
+            Self::GCS(s) => s.download(from, opts, cancel).await,
         }
     }
 
@@ -532,6 +544,7 @@ impl<Other: RemoteStorage> GenericRemoteStorage<Arc<Other>> {
             Self::AwsS3(s) => s.delete(path, cancel).await,
             Self::AzureBlob(s) => s.delete(path, cancel).await,
             Self::Unreliable(s) => s.delete(path, cancel).await,
+            Self::GCS(s) => s.delete(path, cancel).await,
         }
     }
 
@@ -546,6 +559,7 @@ impl<Other: RemoteStorage> GenericRemoteStorage<Arc<Other>> {
             Self::AwsS3(s) => s.delete_objects(paths, cancel).await,
             Self::AzureBlob(s) => s.delete_objects(paths, cancel).await,
             Self::Unreliable(s) => s.delete_objects(paths, cancel).await,
+            Self::GCS(s) => s.delete_objects(paths, cancel).await,
         }
     }
 
@@ -556,6 +570,7 @@ impl<Other: RemoteStorage> GenericRemoteStorage<Arc<Other>> {
             Self::AwsS3(s) => s.max_keys_per_delete(),
             Self::AzureBlob(s) => s.max_keys_per_delete(),
             Self::Unreliable(s) => s.max_keys_per_delete(),
+            Self::GCS(s) => s.max_keys_per_delete(),
         }
     }
 
@@ -570,6 +585,7 @@ impl<Other: RemoteStorage> GenericRemoteStorage<Arc<Other>> {
             Self::AwsS3(s) => s.delete_prefix(prefix, cancel).await,
             Self::AzureBlob(s) => s.delete_prefix(prefix, cancel).await,
             Self::Unreliable(s) => s.delete_prefix(prefix, cancel).await,
+            Self::GCS(_) => todo!(),
         }
     }
 
@@ -585,6 +601,7 @@ impl<Other: RemoteStorage> GenericRemoteStorage<Arc<Other>> {
             Self::AwsS3(s) => s.copy(from, to, cancel).await,
             Self::AzureBlob(s) => s.copy(from, to, cancel).await,
             Self::Unreliable(s) => s.copy(from, to, cancel).await,
+            Self::GCS(_) => todo!(),
         }
     }
 
@@ -613,17 +630,25 @@ impl<Other: RemoteStorage> GenericRemoteStorage<Arc<Other>> {
                 s.time_travel_recover(prefix, timestamp, done_if_after, cancel)
                     .await
             }
+            Self::GCS(_) => todo!(),
         }
     }
 }
 
 impl GenericRemoteStorage {
     pub async fn from_config(storage_config: &RemoteStorageConfig) -> anyhow::Result<Self> {
+        info!("RemoteStorageConfig: {:?}", storage_config);
+
         let timeout = storage_config.timeout;
 
-        // If somkeone overrides timeout to be small without adjusting small_timeout, then adjust it automatically
+        // If someone overrides timeout to be small without adjusting small_timeout, then adjust it automatically
         let small_timeout = std::cmp::min(storage_config.small_timeout, timeout);
 
+        info!(
+            "RemoteStorageConfig's storage attribute: {:?}",
+            storage_config.storage
+        );
+
         Ok(match &storage_config.storage {
             RemoteStorageKind::LocalFs { local_path: path } => {
                 info!("Using fs root '{path}' as a remote storage");
@@ -661,6 +686,16 @@ impl GenericRemoteStorage {
                     small_timeout,
                 )?))
             }
+            RemoteStorageKind::GCS(gcs_config) => {
+                let google_application_credentials =
+                    std::env::var("GOOGLE_APPLICATION_CREDENTIALS")
+                        .unwrap_or_else(|_| "<none>".into());
+                info!(
+                    "Using gcs bucket '{}' as a remote storage, prefix in bucket: '{:?}', GOOGLE_APPLICATION_CREDENTIALS: {google_application_credentials }",
+                    gcs_config.bucket_name, gcs_config.prefix_in_bucket
+                );
+                Self::GCS(Arc::new(GCSBucket::new(gcs_config, timeout).await?))
+            }
         })
     }
 
@@ -690,6 +725,7 @@ impl GenericRemoteStorage {
             Self::AwsS3(s) => Some(s.bucket_name()),
             Self::AzureBlob(s) => Some(s.container_name()),
             Self::Unreliable(_s) => None,
+            Self::GCS(s) => Some(s.bucket_name()),
         }
     }
 }

libs/remote_storage/src/simulate_failures.rs

@@ -50,6 +50,7 @@ impl UnreliableWrapper {
             GenericRemoteStorage::Unreliable(_s) => {
                 panic!("Can't wrap unreliable wrapper unreliably")
             }
+            GenericRemoteStorage::GCS(_) => todo!(),
         };
         UnreliableWrapper {
             inner,

object_storage/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "endpoint_storage"
+name = "object_storage"
 version = "0.0.1"
 edition.workspace = true
 license.workspace = true

object_storage/src/app.rs

@@ -2,7 +2,7 @@ use anyhow::anyhow;
 use axum::body::{Body, Bytes};
 use axum::response::{IntoResponse, Response};
 use axum::{Router, http::StatusCode};
-use endpoint_storage::{PrefixS3Path, S3Path, Storage, bad_request, internal_error, not_found, ok};
+use object_storage::{PrefixS3Path, S3Path, Storage, bad_request, internal_error, not_found, ok};
 use remote_storage::TimeoutOrCancel;
 use remote_storage::{DownloadError, DownloadOpts, GenericRemoteStorage, RemotePath};
 use std::{sync::Arc, time::SystemTime, time::UNIX_EPOCH};
@@ -46,12 +46,12 @@ async fn metrics() -> Result {
 
 async fn get(S3Path { path }: S3Path, state: State) -> Result {
     info!(%path, "downloading");
-    let download_err = |err| {
-        if let DownloadError::NotFound = err {
-            info!(%path, %err, "downloading"); // 404 is not an issue of _this_ service
+    let download_err = |e| {
+        if let DownloadError::NotFound = e {
+            info!(%path, %e, "downloading"); // 404 is not an issue of _this_ service
             return not_found(&path);
         }
-        internal_error(err, &path, "downloading")
+        internal_error(e, &path, "downloading")
     };
     let cancel = state.cancel.clone();
     let opts = &DownloadOpts::default();
@@ -249,7 +249,7 @@ mod tests {
         };
 
         let proxy = Storage {
-            auth: endpoint_storage::JwtAuth::new(TEST_PUB_KEY_ED25519).unwrap(),
+            auth: object_storage::JwtAuth::new(TEST_PUB_KEY_ED25519).unwrap(),
             storage,
             cancel: cancel.clone(),
             max_upload_file_limit: usize::MAX,
@@ -343,14 +343,14 @@ MC4CAQAwBQYDK2VwBCIEID/Drmc1AA6U/znNRWpF3zEGegOATQxfkdWxitcOMsIH
         TimelineId::from_array([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 1, 2, 3, 4, 5, 7]);
     const ENDPOINT_ID: &str = "ep-winter-frost-a662z3vg";
     fn token() -> String {
-        let claims = endpoint_storage::Claims {
+        let claims = object_storage::Claims {
             tenant_id: TENANT_ID,
             timeline_id: TIMELINE_ID,
             endpoint_id: ENDPOINT_ID.into(),
             exp: u64::MAX,
         };
         let key = jsonwebtoken::EncodingKey::from_ed_pem(TEST_PRIV_KEY_ED25519).unwrap();
-        let header = jsonwebtoken::Header::new(endpoint_storage::VALIDATION_ALGO);
+        let header = jsonwebtoken::Header::new(object_storage::VALIDATION_ALGO);
         jsonwebtoken::encode(&header, &claims, &key).unwrap()
     }
 
@@ -364,10 +364,7 @@ MC4CAQAwBQYDK2VwBCIEID/Drmc1AA6U/znNRWpF3zEGegOATQxfkdWxitcOMsIH
             vec![TIMELINE_ID.to_string(), TimelineId::generate().to_string()],
             vec![ENDPOINT_ID, "ep-ololo"]
         )
-        // first one is fully valid path, second path is valid for GET as
-        // read paths may have different endpoint if tenant and timeline matches
-        // (needed for prewarming RO->RW replica)
-        .skip(2);
+        .skip(1);
 
         for ((uri, method), (tenant, timeline, endpoint)) in iproduct!(routes(), args) {
             info!(%uri, %method, %tenant, %timeline, %endpoint);
@@ -478,16 +475,6 @@ MC4CAQAwBQYDK2VwBCIEID/Drmc1AA6U/znNRWpF3zEGegOATQxfkdWxitcOMsIH
         requests_chain(chain.into_iter(), |_| token()).await;
     }
 
-    #[testlog(tokio::test)]
-    async fn read_other_endpoint_data() {
-        let uri = format!("/{TENANT_ID}/{TIMELINE_ID}/other_endpoint/key");
-        let chain = vec![
-            (uri.clone(), "GET", "", StatusCode::NOT_FOUND, false),
-            (uri.clone(), "PUT", "", StatusCode::UNAUTHORIZED, false),
-        ];
-        requests_chain(chain.into_iter(), |_| token()).await;
-    }
-
     fn delete_prefix_token(uri: &str) -> String {
         use serde::Serialize;
         let parts = uri.split("/").collect::<Vec<&str>>();
@@ -495,7 +482,7 @@ MC4CAQAwBQYDK2VwBCIEID/Drmc1AA6U/znNRWpF3zEGegOATQxfkdWxitcOMsIH
         struct PrefixClaims {
             tenant_id: TenantId,
             timeline_id: Option<TimelineId>,
-            endpoint_id: Option<endpoint_storage::EndpointId>,
+            endpoint_id: Option<object_storage::EndpointId>,
             exp: u64,
         }
         let claims = PrefixClaims {
@@ -505,7 +492,7 @@ MC4CAQAwBQYDK2VwBCIEID/Drmc1AA6U/znNRWpF3zEGegOATQxfkdWxitcOMsIH
             exp: u64::MAX,
         };
         let key = jsonwebtoken::EncodingKey::from_ed_pem(TEST_PRIV_KEY_ED25519).unwrap();
-        let header = jsonwebtoken::Header::new(endpoint_storage::VALIDATION_ALGO);
+        let header = jsonwebtoken::Header::new(object_storage::VALIDATION_ALGO);
         jsonwebtoken::encode(&header, &claims, &key).unwrap()
     }
 

object_storage/src/lib.rs

@@ -169,19 +169,10 @@ impl FromRequestParts<Arc<Storage>> for S3Path {
             .auth
             .decode(bearer.token())
             .map_err(|e| bad_request(e, "decoding token"))?;
-
-        // Read paths may have different endpoint ids. For readonly -> readwrite replica
-        // prewarming, endpoint must read other endpoint's data.
-        let endpoint_id = if parts.method == axum::http::Method::GET {
-            claims.endpoint_id.clone()
-        } else {
-            path.endpoint_id.clone()
-        };
-
         let route = Claims {
             tenant_id: path.tenant_id,
             timeline_id: path.timeline_id,
-            endpoint_id,
+            endpoint_id: path.endpoint_id.clone(),
             exp: claims.exp,
         };
         if route != claims {

object_storage/src/main.rs

@@ -1,4 +1,4 @@
-//! `endpoint_storage` is a service which provides API for uploading and downloading
+//! `object_storage` is a service which provides API for uploading and downloading
 //! files. It is used by compute and control plane for accessing LFC prewarm data.
 //! This service is deployed either as a separate component or as part of compute image
 //! for large computes.
@@ -33,7 +33,7 @@ async fn main() -> anyhow::Result<()> {
 
     let config: String = std::env::args().skip(1).take(1).collect();
     if config.is_empty() {
-        anyhow::bail!("Usage: endpoint_storage config.json")
+        anyhow::bail!("Usage: object_storage config.json")
     }
     info!("Reading config from {config}");
     let config = std::fs::read_to_string(config.clone())?;
@@ -41,7 +41,7 @@ async fn main() -> anyhow::Result<()> {
     info!("Reading pemfile from {}", config.pemfile.clone());
     let pemfile = std::fs::read(config.pemfile.clone())?;
     info!("Loading public key from {}", config.pemfile.clone());
-    let auth = endpoint_storage::JwtAuth::new(&pemfile)?;
+    let auth = object_storage::JwtAuth::new(&pemfile)?;
 
     let listener = tokio::net::TcpListener::bind(config.listen).await.unwrap();
     info!("listening on {}", listener.local_addr().unwrap());
@@ -50,7 +50,7 @@ async fn main() -> anyhow::Result<()> {
     let cancel = tokio_util::sync::CancellationToken::new();
     app::check_storage_permissions(&storage, cancel.clone()).await?;
 
-    let proxy = std::sync::Arc::new(endpoint_storage::Storage {
+    let proxy = std::sync::Arc::new(object_storage::Storage {
         auth,
         storage,
         cancel: cancel.clone(),

pageserver/src/consumption_metrics/metrics.rs

@@ -263,9 +263,7 @@ where
     while let Some((tenant_id, tenant)) = tenants.next().await {
         let mut tenant_resident_size = 0;
 
-        let timelines = tenant.list_timelines();
-        let timelines_len = timelines.len();
-        for timeline in timelines {
+        for timeline in tenant.list_timelines() {
             let timeline_id = timeline.timeline_id;
 
             match TimelineSnapshot::collect(&timeline, ctx) {
@@ -291,11 +289,6 @@ where
             tenant_resident_size += timeline.resident_physical_size();
         }
 
-        if timelines_len == 0 {
-            // Force set it to 1 byte to avoid not being reported -- all timelines are offloaded.
-            tenant_resident_size = 1;
-        }
-
         let snap = TenantSnapshot::collect(&tenant, tenant_resident_size);
         snap.to_metrics(tenant_id, Utc::now(), cache, &mut current_metrics);
     }

pageserver/src/tenant/timeline.rs

@@ -1285,10 +1285,6 @@ impl Timeline {
         reconstruct_state: &mut ValuesReconstructState,
         ctx: &RequestContext,
     ) -> Result<BTreeMap<Key, Result<Bytes, PageReconstructError>>, GetVectoredError> {
-        if query.is_empty() {
-            return Ok(BTreeMap::default());
-        }
-
         let read_path = if self.conf.enable_read_path_debugging || ctx.read_path_debug() {
             Some(ReadPath::new(
                 query.total_keyspace(),

pgxn/neon/pagestore_smgr.c

@@ -803,13 +803,7 @@ neon_create(SMgrRelation reln, ForkNumber forkNum, bool isRedo)
 
 		case RELPERSISTENCE_TEMP:
 		case RELPERSISTENCE_UNLOGGED:
-#ifdef DEBUG_COMPARE_LOCAL
-			mdcreate(reln, forkNum, forkNum == INIT_FORKNUM || isRedo);
-			if (forkNum == MAIN_FORKNUM)
-				mdcreate(reln, INIT_FORKNUM, true);
-#else
 			mdcreate(reln, forkNum, isRedo);
-#endif
 			return;
 
 		default:
@@ -1979,10 +1973,6 @@ neon_start_unlogged_build(SMgrRelation reln)
 		case RELPERSISTENCE_UNLOGGED:
 			unlogged_build_rel = reln;
 			unlogged_build_phase = UNLOGGED_BUILD_NOT_PERMANENT;
-#ifdef DEBUG_COMPARE_LOCAL
-			if (!IsParallelWorker())
-				mdcreate(reln, INIT_FORKNUM, true);
-#endif
 			return;
 
 		default:
@@ -2005,14 +1995,12 @@ neon_start_unlogged_build(SMgrRelation reln)
 	 * FIXME: should we pass isRedo true to create the tablespace dir if it
 	 * doesn't exist? Is it needed?
 	 */
- 	if (!IsParallelWorker())
-	{
 #ifndef DEBUG_COMPARE_LOCAL
+ 	if (!IsParallelWorker())
 		mdcreate(reln, MAIN_FORKNUM, false);
 #else
-		mdcreate(reln, INIT_FORKNUM, true);
+	mdcreate(reln, INIT_FORKNUM, false);
 #endif
-	}
 }
 
 /*
@@ -2111,12 +2099,12 @@ neon_end_unlogged_build(SMgrRelation reln)
 #ifndef DEBUG_COMPARE_LOCAL
 			/* use isRedo == true, so that we drop it immediately */
 			mdunlink(rinfob, forknum, true);
+#else
+			mdunlink(rinfob, INIT_FORKNUM, true);
 #endif
 		}
-#ifdef DEBUG_COMPARE_LOCAL
-		mdunlink(rinfob, INIT_FORKNUM, true);
-#endif
 	}
+
 	unlogged_build_rel = NULL;
 	unlogged_build_phase = UNLOGGED_BUILD_NOT_IN_PROGRESS;
 }

proxy/src/lib.rs

@@ -91,7 +91,6 @@ mod jemalloc;
 mod logging;
 mod metrics;
 mod parse;
-mod pglb;
 mod protocol2;
 mod proxy;
 mod rate_limiter;

proxy/src/pglb/inprocess.rs

@@ -1,193 +0,0 @@
-#![allow(dead_code, reason = "TODO: work in progress")]
-
-use std::pin::{Pin, pin};
-use std::sync::Arc;
-use std::sync::atomic::{AtomicUsize, Ordering};
-use std::task::{Context, Poll};
-use std::{fmt, io};
-
-use tokio::io::{AsyncRead, AsyncWrite, DuplexStream, ReadBuf};
-use tokio::sync::mpsc;
-
-const STREAM_CHANNEL_SIZE: usize = 16;
-const MAX_STREAM_BUFFER_SIZE: usize = 4096;
-
-#[derive(Debug)]
-pub struct Connection {
-    stream_sender: mpsc::Sender<Stream>,
-    stream_receiver: mpsc::Receiver<Stream>,
-    stream_id_counter: Arc<AtomicUsize>,
-}
-
-impl Connection {
-    pub fn new() -> (Connection, Connection) {
-        let (sender_a, receiver_a) = mpsc::channel(STREAM_CHANNEL_SIZE);
-        let (sender_b, receiver_b) = mpsc::channel(STREAM_CHANNEL_SIZE);
-
-        let stream_id_counter = Arc::new(AtomicUsize::new(1));
-
-        let conn_a = Connection {
-            stream_sender: sender_a,
-            stream_receiver: receiver_b,
-            stream_id_counter: Arc::clone(&stream_id_counter),
-        };
-        let conn_b = Connection {
-            stream_sender: sender_b,
-            stream_receiver: receiver_a,
-            stream_id_counter,
-        };
-
-        (conn_a, conn_b)
-    }
-
-    #[inline]
-    fn next_stream_id(&self) -> StreamId {
-        StreamId(self.stream_id_counter.fetch_add(1, Ordering::Relaxed))
-    }
-
-    #[tracing::instrument(skip_all, fields(stream_id = tracing::field::Empty, err))]
-    pub async fn open_stream(&self) -> io::Result<Stream> {
-        let (local, remote) = tokio::io::duplex(MAX_STREAM_BUFFER_SIZE);
-        let stream_id = self.next_stream_id();
-        tracing::Span::current().record("stream_id", stream_id.0);
-
-        let local = Stream {
-            inner: local,
-            id: stream_id,
-        };
-        let remote = Stream {
-            inner: remote,
-            id: stream_id,
-        };
-
-        self.stream_sender
-            .send(remote)
-            .await
-            .map_err(io::Error::other)?;
-
-        Ok(local)
-    }
-
-    #[tracing::instrument(skip_all, fields(stream_id = tracing::field::Empty, err))]
-    pub async fn accept_stream(&mut self) -> io::Result<Option<Stream>> {
-        Ok(self.stream_receiver.recv().await.inspect(|stream| {
-            tracing::Span::current().record("stream_id", stream.id.0);
-        }))
-    }
-}
-
-#[derive(Copy, Clone, Debug)]
-pub struct StreamId(usize);
-
-impl fmt::Display for StreamId {
-    #[inline]
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(f, "{}", self.0)
-    }
-}
-
-// TODO: Proper closing. Currently Streams can outlive their Connections.
-// Carry WeakSender and check strong_count?
-#[derive(Debug)]
-pub struct Stream {
-    inner: DuplexStream,
-    id: StreamId,
-}
-
-impl Stream {
-    #[inline]
-    pub fn id(&self) -> StreamId {
-        self.id
-    }
-}
-
-impl AsyncRead for Stream {
-    #[tracing::instrument(level = "debug", skip_all, fields(stream_id = %self.id))]
-    #[inline]
-    fn poll_read(
-        mut self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-        buf: &mut ReadBuf<'_>,
-    ) -> Poll<io::Result<()>> {
-        pin!(&mut self.inner).poll_read(cx, buf)
-    }
-}
-
-impl AsyncWrite for Stream {
-    #[tracing::instrument(level = "debug", skip_all, fields(stream_id = %self.id))]
-    #[inline]
-    fn poll_write(
-        mut self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-        buf: &[u8],
-    ) -> Poll<Result<usize, io::Error>> {
-        pin!(&mut self.inner).poll_write(cx, buf)
-    }
-
-    #[tracing::instrument(level = "debug", skip_all, fields(stream_id = %self.id))]
-    #[inline]
-    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
-        pin!(&mut self.inner).poll_flush(cx)
-    }
-
-    #[tracing::instrument(level = "debug", skip_all, fields(stream_id = %self.id))]
-    #[inline]
-    fn poll_shutdown(
-        mut self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Result<(), io::Error>> {
-        pin!(&mut self.inner).poll_shutdown(cx)
-    }
-
-    #[tracing::instrument(level = "debug", skip_all, fields(stream_id = %self.id))]
-    #[inline]
-    fn poll_write_vectored(
-        mut self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-        bufs: &[io::IoSlice<'_>],
-    ) -> Poll<Result<usize, io::Error>> {
-        pin!(&mut self.inner).poll_write_vectored(cx, bufs)
-    }
-
-    #[inline]
-    fn is_write_vectored(&self) -> bool {
-        self.inner.is_write_vectored()
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use tokio::io::{AsyncReadExt, AsyncWriteExt};
-
-    use super::*;
-
-    #[tokio::test]
-    async fn test_simple_roundtrip() {
-        let (client, mut server) = Connection::new();
-
-        let server_task = tokio::spawn(async move {
-            while let Some(mut stream) = server.accept_stream().await.unwrap() {
-                tokio::spawn(async move {
-                    let mut buf = [0; 64];
-                    loop {
-                        match stream.read(&mut buf).await.unwrap() {
-                            0 => break,
-                            n => stream.write(&buf[..n]).await.unwrap(),
-                        };
-                    }
-                });
-            }
-        });
-
-        let mut stream = client.open_stream().await.unwrap();
-        stream.write_all(b"hello!").await.unwrap();
-        let mut buf = [0; 64];
-        let n = stream.read(&mut buf).await.unwrap();
-        assert_eq!(n, 6);
-        assert_eq!(&buf[..n], b"hello!");
-
-        drop(stream);
-        drop(client);
-        server_task.await.unwrap();
-    }
-}

proxy/src/pglb/mod.rs

@@ -1 +0,0 @@
-pub mod inprocess;

safekeeper/src/bin/safekeeper.rs

@@ -14,7 +14,6 @@ use clap::{ArgAction, Parser};
 use futures::future::BoxFuture;
 use futures::stream::FuturesUnordered;
 use futures::{FutureExt, StreamExt};
-use http_utils::tls_certs::ReloadingCertificateResolver;
 use metrics::set_build_info_metric;
 use remote_storage::RemoteStorageConfig;
 use safekeeper::defaults::{
@@ -24,8 +23,8 @@ use safekeeper::defaults::{
     DEFAULT_SSL_CERT_RELOAD_PERIOD, DEFAULT_SSL_KEY_FILE,
 };
 use safekeeper::{
-    BACKGROUND_RUNTIME, BROKER_RUNTIME, GlobalTimelines, HTTP_RUNTIME, SafeKeeperConf,
-    WAL_SERVICE_RUNTIME, broker, control_file, http, wal_backup, wal_service,
+    BROKER_RUNTIME, GlobalTimelines, HTTP_RUNTIME, SafeKeeperConf, WAL_SERVICE_RUNTIME, broker,
+    control_file, http, wal_backup, wal_service,
 };
 use sd_notify::NotifyState;
 use storage_broker::{DEFAULT_ENDPOINT, Uri};
@@ -216,21 +215,16 @@ struct Args {
     ssl_cert_file: Utf8PathBuf,
     /// Period to reload certificate and private key from files.
     #[arg(long, value_parser = humantime::parse_duration, default_value = DEFAULT_SSL_CERT_RELOAD_PERIOD)]
-    ssl_cert_reload_period: Duration,
+    pub ssl_cert_reload_period: Duration,
     /// Trusted root CA certificates to use in https APIs.
     #[arg(long)]
-    ssl_ca_file: Option<Utf8PathBuf>,
+    pub ssl_ca_file: Option<Utf8PathBuf>,
     /// Flag to use https for requests to peer's safekeeper API.
     #[arg(long)]
-    use_https_safekeeper_api: bool,
+    pub use_https_safekeeper_api: bool,
     /// Path to the JWT auth token used to authenticate with other safekeepers.
     #[arg(long)]
     auth_token_path: Option<Utf8PathBuf>,
-    /// Enable TLS in WAL service API.
-    /// Does not force TLS: the client negotiates TLS usage during the handshake.
-    /// Uses key and certificate from ssl_key_file/ssl_cert_file.
-    #[arg(long)]
-    enable_tls_wal_service_api: bool,
 }
 
 // Like PathBufValueParser, but allows empty string.
@@ -424,7 +418,6 @@ async fn main() -> anyhow::Result<()> {
         ssl_cert_reload_period: args.ssl_cert_reload_period,
         ssl_ca_certs,
         use_https_safekeeper_api: args.use_https_safekeeper_api,
-        enable_tls_wal_service_api: args.enable_tls_wal_service_api,
     });
 
     // initialize sentry if SENTRY_DSN is provided
@@ -524,36 +517,6 @@ async fn start_safekeeper(conf: Arc<SafeKeeperConf>) -> Result<()> {
         info!("running in current thread runtime");
     }
 
-    let tls_server_config = if conf.listen_https_addr.is_some() || conf.enable_tls_wal_service_api {
-        let ssl_key_file = conf.ssl_key_file.clone();
-        let ssl_cert_file = conf.ssl_cert_file.clone();
-        let ssl_cert_reload_period = conf.ssl_cert_reload_period;
-
-        // Create resolver in BACKGROUND_RUNTIME, so the background certificate reloading
-        // task is run in this runtime.
-        let cert_resolver = current_thread_rt
-            .as_ref()
-            .unwrap_or_else(|| BACKGROUND_RUNTIME.handle())
-            .spawn(async move {
-                ReloadingCertificateResolver::new(
-                    "main",
-                    &ssl_key_file,
-                    &ssl_cert_file,
-                    ssl_cert_reload_period,
-                )
-                .await
-            })
-            .await??;
-
-        let config = rustls::ServerConfig::builder()
-            .with_no_client_auth()
-            .with_cert_resolver(cert_resolver);
-
-        Some(Arc::new(config))
-    } else {
-        None
-    };
-
     let wal_service_handle = current_thread_rt
         .as_ref()
         .unwrap_or_else(|| WAL_SERVICE_RUNTIME.handle())
@@ -561,9 +524,6 @@ async fn start_safekeeper(conf: Arc<SafeKeeperConf>) -> Result<()> {
             conf.clone(),
             pg_listener,
             Scope::SafekeeperData,
-            conf.enable_tls_wal_service_api
-                .then(|| tls_server_config.clone())
-                .flatten(),
             global_timelines.clone(),
         ))
         // wrap with task name for error reporting
@@ -592,9 +552,6 @@ async fn start_safekeeper(conf: Arc<SafeKeeperConf>) -> Result<()> {
                 conf.clone(),
                 pg_listener_tenant_only,
                 Scope::Tenant,
-                conf.enable_tls_wal_service_api
-                    .then(|| tls_server_config.clone())
-                    .flatten(),
                 global_timelines.clone(),
             ))
             // wrap with task name for error reporting
@@ -620,7 +577,6 @@ async fn start_safekeeper(conf: Arc<SafeKeeperConf>) -> Result<()> {
             .spawn(http::task_main_https(
                 conf.clone(),
                 https_listener,
-                tls_server_config.expect("tls_server_config is set earlier if https is enabled"),
                 global_timelines.clone(),
             ))
             .map(|res| ("HTTPS service main".to_owned(), res));

safekeeper/src/http/mod.rs

@@ -1,6 +1,7 @@
 pub mod routes;
 use std::sync::Arc;
 
+use http_utils::tls_certs::ReloadingCertificateResolver;
 pub use routes::make_router;
 pub use safekeeper_api::models;
 use tokio_util::sync::CancellationToken;
@@ -27,10 +28,21 @@ pub async fn task_main_http(
 pub async fn task_main_https(
     conf: Arc<SafeKeeperConf>,
     https_listener: std::net::TcpListener,
-    tls_config: Arc<rustls::ServerConfig>,
     global_timelines: Arc<GlobalTimelines>,
 ) -> anyhow::Result<()> {
-    let tls_acceptor = tokio_rustls::TlsAcceptor::from(tls_config);
+    let cert_resolver = ReloadingCertificateResolver::new(
+        "main",
+        &conf.ssl_key_file,
+        &conf.ssl_cert_file,
+        conf.ssl_cert_reload_period,
+    )
+    .await?;
+
+    let server_config = rustls::ServerConfig::builder()
+        .with_no_client_auth()
+        .with_cert_resolver(cert_resolver);
+
+    let tls_acceptor = tokio_rustls::TlsAcceptor::from(Arc::new(server_config));
 
     let router = make_router(conf, global_timelines)
         .build()

safekeeper/src/lib.rs

@@ -122,7 +122,6 @@ pub struct SafeKeeperConf {
     pub ssl_cert_reload_period: Duration,
     pub ssl_ca_certs: Vec<Pem>,
     pub use_https_safekeeper_api: bool,
-    pub enable_tls_wal_service_api: bool,
 }
 
 impl SafeKeeperConf {
@@ -173,7 +172,6 @@ impl SafeKeeperConf {
             ssl_cert_reload_period: Duration::from_secs(60),
             ssl_ca_certs: Vec::new(),
             use_https_safekeeper_api: false,
-            enable_tls_wal_service_api: false,
         }
     }
 }
@@ -211,12 +209,3 @@ pub static WAL_BACKUP_RUNTIME: Lazy<Runtime> = Lazy::new(|| {
         .build()
         .expect("Failed to create WAL backup runtime")
 });
-
-pub static BACKGROUND_RUNTIME: Lazy<Runtime> = Lazy::new(|| {
-    tokio::runtime::Builder::new_multi_thread()
-        .thread_name("background worker")
-        .worker_threads(1) // there is only one task now (ssl certificate reloading), having more threads doesn't make sense
-        .enable_all()
-        .build()
-        .expect("Failed to create background runtime")
-});

safekeeper/src/wal_service.rs

@@ -29,7 +29,6 @@ pub async fn task_main(
     conf: Arc<SafeKeeperConf>,
     pg_listener: std::net::TcpListener,
     allowed_auth_scope: Scope,
-    tls_config: Option<Arc<rustls::ServerConfig>>,
     global_timelines: Arc<GlobalTimelines>,
 ) -> anyhow::Result<()> {
     // Tokio's from_std won't do this for us, per its comment.
@@ -44,10 +43,9 @@ pub async fn task_main(
         let conf = conf.clone();
         let conn_id = issue_connection_id(&mut connection_count);
         let global_timelines = global_timelines.clone();
-        let tls_config = tls_config.clone();
         tokio::spawn(
             async move {
-                if let Err(err) = handle_socket(socket, conf, conn_id, allowed_auth_scope, tls_config, global_timelines).await {
+                if let Err(err) = handle_socket(socket, conf, conn_id, allowed_auth_scope, global_timelines).await {
                     error!("connection handler exited: {}", err);
                 }
             }
@@ -63,7 +61,6 @@ async fn handle_socket(
     conf: Arc<SafeKeeperConf>,
     conn_id: ConnectionId,
     allowed_auth_scope: Scope,
-    tls_config: Option<Arc<rustls::ServerConfig>>,
     global_timelines: Arc<GlobalTimelines>,
 ) -> Result<(), QueryError> {
     socket.set_nodelay(true)?;
@@ -113,8 +110,7 @@ async fn handle_socket(
         auth_pair,
         global_timelines,
     );
-    let pgbackend =
-        PostgresBackend::new_from_io(socket_fd, socket, peer_addr, auth_type, tls_config)?;
+    let pgbackend = PostgresBackend::new_from_io(socket_fd, socket, peer_addr, auth_type, None)?;
     // libpq protocol between safekeeper and walproposer / pageserver
     // We don't use shutdown.
     pgbackend

safekeeper/tests/walproposer_sim/safekeeper.rs

@@ -185,7 +185,6 @@ pub fn run_server(os: NodeOs, disk: Arc<SafekeeperDisk>) -> Result<()> {
         ssl_cert_reload_period: Duration::ZERO,
         ssl_ca_certs: Vec::new(),
         use_https_safekeeper_api: false,
-        enable_tls_wal_service_api: false,
     };
 
     let mut global = GlobalMap::new(disk, conf.clone())?;

storage_controller/src/service/safekeeper_service.rs

@@ -151,39 +151,11 @@ impl Service {
             "Got {} non-successful responses from initial creation request of total {total_result_count} responses",
             remaining.len()
         );
-        let target_sk_count = timeline_persistence.sk_set.len();
-        let quorum_size = match target_sk_count {
-            0 => {
-                return Err(ApiError::InternalServerError(anyhow::anyhow!(
-                    "timeline configured without any safekeepers",
-                )));
-            }
-            1 | 2 => {
-                #[cfg(feature = "testing")]
-                {
-                    // In test settings, it is allowed to have one or two safekeepers
-                    target_sk_count
-                }
-                #[cfg(not(feature = "testing"))]
-                {
-                    // The region is misconfigured: we need at least three safekeepers to be configured
-                    // in order to schedule work to them
-                    tracing::warn!(
-                        "couldn't find at least 3 safekeepers for timeline, found: {:?}",
-                        timeline_persistence.sk_set
-                    );
-                    return Err(ApiError::InternalServerError(anyhow::anyhow!(
-                        "couldn't find at least 3 safekeepers to put timeline to"
-                    )));
-                }
-            }
-            _ => target_sk_count / 2 + 1,
-        };
-        let success_count = target_sk_count - remaining.len();
-        if success_count < quorum_size {
+        if remaining.len() >= 2 {
             // Failure
             return Err(ApiError::InternalServerError(anyhow::anyhow!(
-                "not enough successful reconciliations to reach quorum size: {success_count} of {quorum_size} of total {target_sk_count}"
+                "not enough successful reconciliations to reach quorum, please retry: {} errored",
+                remaining.len()
             )));
         }
 
@@ -520,6 +492,8 @@ impl Service {
     pub(crate) async fn safekeepers_for_new_timeline(
         &self,
     ) -> Result<Vec<SafekeeperInfo>, ApiError> {
+        // Number of safekeepers in different AZs we are looking for
+        let wanted_count = 3;
         let mut all_safekeepers = {
             let locked = self.inner.read().unwrap();
             locked
@@ -558,19 +532,6 @@ impl Service {
                 sk.1.id.0,
             )
         });
-        // Number of safekeepers in different AZs we are looking for
-        let wanted_count = match all_safekeepers.len() {
-            0 => {
-                return Err(ApiError::InternalServerError(anyhow::anyhow!(
-                    "couldn't find any active safekeeper for new timeline",
-                )));
-            }
-            // Have laxer requirements on testig mode as we don't want to
-            // spin up three safekeepers for every single test
-            #[cfg(feature = "testing")]
-            1 | 2 => all_safekeepers.len(),
-            _ => 3,
-        };
         let mut sks = Vec::new();
         let mut azs = HashSet::new();
         for (_sk_util, sk_info, az_id) in all_safekeepers.iter() {

storage_scrubber/src/lib.rs

@@ -271,6 +271,7 @@ impl BucketConfig {
                 "container {}, storage account {:?}, region {}",
                 config.container_name, config.storage_account, config.container_region
             ),
+            RemoteStorageKind::GCS(config) => format!("bucket {}", config.bucket_name),
         }
     }
     pub fn bucket_name(&self) -> Option<&str> {
@@ -418,6 +419,9 @@ async fn init_remote(
             config.prefix_in_container.get_or_insert(default_prefix);
         }
         RemoteStorageKind::LocalFs { .. } => (),
+        RemoteStorageKind::GCS(config) => {
+            config.prefix_in_bucket.get_or_insert(default_prefix);
+        }
     }
 
     // We already pass the prefix to the remote client above

test_runner/fixtures/neon_cli.py

@@ -417,14 +417,14 @@ class NeonLocalCli(AbstractNeonCli):
             cmd.append(f"--instance-id={instance_id}")
         return self.raw_cli(cmd)
 
-    def endpoint_storage_start(self, timeout_in_seconds: int | None = None):
-        cmd = ["endpoint-storage", "start"]
+    def object_storage_start(self, timeout_in_seconds: int | None = None):
+        cmd = ["object-storage", "start"]
         if timeout_in_seconds is not None:
             cmd.append(f"--start-timeout={timeout_in_seconds}s")
         return self.raw_cli(cmd)
 
-    def endpoint_storage_stop(self, immediate: bool):
-        cmd = ["endpoint-storage", "stop"]
+    def object_storage_stop(self, immediate: bool):
+        cmd = ["object-storage", "stop"]
         if immediate:
             cmd.extend(["-m", "immediate"])
         return self.raw_cli(cmd)

test_runner/fixtures/neon_fixtures.py

@@ -1029,7 +1029,7 @@ class NeonEnvBuilder:
 
             self.env.broker.assert_no_errors()
 
-            self.env.endpoint_storage.assert_no_errors()
+            self.env.object_storage.assert_no_errors()
 
         try:
             self.overlay_cleanup_teardown()
@@ -1126,7 +1126,7 @@ class NeonEnv:
             pagectl_env_vars["RUST_LOG"] = self.rust_log_override
         self.pagectl = Pagectl(extra_env=pagectl_env_vars, binpath=self.neon_binpath)
 
-        self.endpoint_storage = EndpointStorage(self)
+        self.object_storage = ObjectStorage(self)
 
         # The URL for the pageserver to use as its control_plane_api config
         if config.storage_controller_port_override is not None:
@@ -1183,7 +1183,7 @@ class NeonEnv:
             },
             "safekeepers": [],
             "pageservers": [],
-            "endpoint_storage": {"port": self.port_distributor.get_port()},
+            "object_storage": {"port": self.port_distributor.get_port()},
             "generate_local_ssl_certs": self.generate_local_ssl_certs,
         }
 
@@ -1420,7 +1420,7 @@ class NeonEnv:
                 self.storage_controller.on_safekeeper_deploy(sk_id, body)
                 self.storage_controller.safekeeper_scheduling_policy(sk_id, "Active")
 
-        self.endpoint_storage.start(timeout_in_seconds=timeout_in_seconds)
+        self.object_storage.start(timeout_in_seconds=timeout_in_seconds)
 
     def stop(self, immediate=False, ps_assert_metric_no_errors=False, fail_on_endpoint_errors=True):
         """
@@ -1439,7 +1439,7 @@ class NeonEnv:
         except Exception as e:
             raise_later = e
 
-        self.endpoint_storage.stop(immediate=immediate)
+        self.object_storage.stop(immediate=immediate)
 
         # Stop storage controller before pageservers: we don't want it to spuriously
         # detect a pageserver "failure" during test teardown
@@ -2660,24 +2660,24 @@ class NeonStorageController(MetricsGetter, LogUtils):
         self.stop(immediate=True)
 
 
-class EndpointStorage(LogUtils):
+class ObjectStorage(LogUtils):
     def __init__(self, env: NeonEnv):
-        service_dir = env.repo_dir / "endpoint_storage"
-        super().__init__(logfile=service_dir / "endpoint_storage.log")
-        self.conf_path = service_dir / "endpoint_storage.json"
+        service_dir = env.repo_dir / "object_storage"
+        super().__init__(logfile=service_dir / "object_storage.log")
+        self.conf_path = service_dir / "object_storage.json"
         self.env = env
 
     def base_url(self):
         return json.loads(self.conf_path.read_text())["listen"]
 
     def start(self, timeout_in_seconds: int | None = None):
-        self.env.neon_cli.endpoint_storage_start(timeout_in_seconds)
+        self.env.neon_cli.object_storage_start(timeout_in_seconds)
 
     def stop(self, immediate: bool = False):
-        self.env.neon_cli.endpoint_storage_stop(immediate)
+        self.env.neon_cli.object_storage_stop(immediate)
 
     def assert_no_errors(self):
-        assert_no_errors(self.logfile, "endpoint_storage", [])
+        assert_no_errors(self.logfile, "object_storage", [])
 
 
 class NeonProxiedStorageController(NeonStorageController):

test_runner/performance/test_physical_replication.py

@@ -65,7 +65,7 @@ def test_ro_replica_lag(
     project = neon_api.create_project(pg_version)
     project_id = project["project"]["id"]
     log.info("Project ID: %s", project_id)
-    log.info("Primary endpoint ID: %s", project["endpoints"][0]["id"])
+    log.info("Primary endpoint ID: %s", project["project"]["endpoints"][0]["id"])
     neon_api.wait_for_operation_to_finish(project_id)
     error_occurred = False
     try:
@@ -198,7 +198,7 @@ def test_replication_start_stop(
     project = neon_api.create_project(pg_version)
     project_id = project["project"]["id"]
     log.info("Project ID: %s", project_id)
-    log.info("Primary endpoint ID: %s", project["endpoints"][0]["id"])
+    log.info("Primary endpoint ID: %s", project["project"]["endpoints"][0]["id"])
     neon_api.wait_for_operation_to_finish(project_id)
     try:
         branch_id = project["branch"]["id"]

test_runner/regress/test_download_extensions.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import os
-import platform
 import shutil
 import tarfile
 from typing import TYPE_CHECKING
@@ -59,18 +58,7 @@ def test_remote_extensions(
     extensions_endpoint = f"http://{host}:{port}/pg-ext-s3-gateway"
 
     build_tag = os.environ.get("BUILD_TAG", "latest")
-
-    # We have decided to use the Go naming convention due to Kubernetes.
-    arch = platform.machine()
-    match arch:
-        case "aarch64":
-            arch = "arm64"
-        case "x86_64":
-            arch = "amd64"
-        case _:
-            pass
-
-    archive_route = f"{build_tag}/{arch}/v{pg_version}/extensions/test_extension.tar.zst"
+    archive_route = f"{build_tag}/v{pg_version}/extensions/test_extension.tar.zst"
     tarball = test_output_dir / "test_extension.tar"
     extension_dir = (
         base_dir / "test_runner" / "regress" / "data" / "test_remote_extensions" / "test_extension"

test_runner/regress/test_neon_cli.py

@@ -138,7 +138,7 @@ def test_cli_start_stop(neon_env_builder: NeonEnvBuilder):
     env.neon_cli.pageserver_stop(env.pageserver.id)
     env.neon_cli.safekeeper_stop()
     env.neon_cli.storage_controller_stop(False)
-    env.neon_cli.endpoint_storage_stop(False)
+    env.neon_cli.object_storage_stop(False)
     env.neon_cli.storage_broker_stop()
 
     # Keep NeonEnv state up to date, it usually owns starting/stopping services
@@ -185,7 +185,7 @@ def test_cli_start_stop_multi(neon_env_builder: NeonEnvBuilder):
     env.neon_cli.safekeeper_stop(neon_env_builder.safekeepers_id_start + 1)
     env.neon_cli.safekeeper_stop(neon_env_builder.safekeepers_id_start + 2)
 
-    env.neon_cli.endpoint_storage_stop(False)
+    env.neon_cli.object_storage_stop(False)
 
     # Stop this to get out of the way of the following `start`
     env.neon_cli.storage_controller_stop(False)

test_runner/regress/test_object_storage.py

@@ -8,7 +8,7 @@ from jwcrypto import jwk, jwt
 
 
 @pytest.mark.asyncio
-async def test_endpoint_storage_insert_retrieve_delete(neon_simple_env: NeonEnv):
+async def test_object_storage_insert_retrieve_delete(neon_simple_env: NeonEnv):
     """
     Inserts, retrieves, and deletes test file using a JWT token
     """
@@ -31,7 +31,7 @@ async def test_endpoint_storage_insert_retrieve_delete(neon_simple_env: NeonEnv)
     token.make_signed_token(key)
     token = token.serialize()
 
-    base_url = env.endpoint_storage.base_url()
+    base_url = env.object_storage.base_url()
     key = f"http://{base_url}/{tenant_id}/{timeline_id}/{endpoint_id}/key"
     headers = {"Authorization": f"Bearer {token}"}
     log.info(f"cache key url {key}")

test_runner/regress/test_storage_controller.py

@@ -95,7 +95,7 @@ def test_storage_controller_smoke(
     env.pageservers[1].start()
     for sk in env.safekeepers:
         sk.start()
-    env.endpoint_storage.start()
+    env.object_storage.start()
 
     # The pageservers we started should have registered with the sharding service on startup
     nodes = env.storage_controller.node_list()
@@ -347,7 +347,7 @@ def prepare_onboarding_env(
     env = neon_env_builder.init_configs()
     env.broker.start()
     env.storage_controller.start()
-    env.endpoint_storage.start()
+    env.object_storage.start()
 
     # This is the pageserver where we'll initially create the tenant.  Run it in emergency
     # mode so that it doesn't talk to storage controller, and do not register it.
@@ -1612,18 +1612,16 @@ def test_storage_controller_heartbeats(
     env = neon_env_builder.init_configs()
     env.start()
 
-    env.storage_controller.allowed_errors.extend(
-        [
-            # Default log allow list permits connection errors, but this test will use error responses on
-            # the utilization endpoint.
-            ".*Call to node.*management API.*failed.*failpoint.*",
-            # The server starts listening to the socket before sending re-attach request,
-            # but it starts serving HTTP only when re-attach is completed.
-            # If re-attach is slow (last scenario), storcon's heartbeat requests will time out.
-            ".*Call to node.*management API.*failed.* Timeout.*",
-            # We will intentionally cause reconcile errors
-            ".*Reconcile error.*",
-        ]
+    # Default log allow list permits connection errors, but this test will use error responses on
+    # the utilization endpoint.
+    env.storage_controller.allowed_errors.append(
+        ".*Call to node.*management API.*failed.*failpoint.*"
+    )
+    # The server starts listening to the socket before sending re-attach request,
+    # but it starts serving HTTP only when re-attach is completed.
+    # If re-attach is slow (last scenario), storcon's heartbeat requests will time out.
+    env.storage_controller.allowed_errors.append(
+        ".*Call to node.*management API.*failed.* Timeout.*"
     )
 
     # Initially we have two online pageservers
@@ -4242,63 +4240,6 @@ def test_storcon_create_delete_sk_down(
     wait_until(timeline_deleted_on_sk)
 
 
-@run_only_on_default_postgres("PG version is not interesting here")
-@pytest.mark.parametrize("num_safekeepers", [1, 2, 3])
-@pytest.mark.parametrize("deletetion_subject", [DeletionSubject.TENANT, DeletionSubject.TIMELINE])
-def test_storcon_few_sk(
-    neon_env_builder: NeonEnvBuilder,
-    num_safekeepers: int,
-    deletetion_subject: DeletionSubject,
-):
-    """
-    Test that the storcon can create and delete tenants and timelines with a limited/special number of safekeepers
-      - num_safekeepers: number of safekeepers.
-      - deletion_subject: test that both single timeline and whole tenant deletion work.
-    """
-
-    neon_env_builder.num_safekeepers = num_safekeepers
-    safekeeper_list = list(range(1, num_safekeepers + 1))
-    neon_env_builder.storage_controller_config = {
-        "timelines_onto_safekeepers": True,
-    }
-    env = neon_env_builder.init_start()
-
-    tenant_id = TenantId.generate()
-    timeline_id = TimelineId.generate()
-    env.create_tenant(tenant_id, timeline_id)
-    child_timeline_id = env.create_branch("child_of_main", tenant_id)
-
-    env.safekeepers[0].assert_log_contains(f"creating new timeline {tenant_id}/{timeline_id}")
-
-    config_lines = [
-        "neon.safekeeper_proto_version = 3",
-    ]
-    with env.endpoints.create("main", tenant_id=tenant_id, config_lines=config_lines) as ep:
-        # endpoint should start.
-        ep.start(safekeeper_generation=1, safekeepers=safekeeper_list)
-        ep.safe_psql("CREATE TABLE IF NOT EXISTS t(key int, value text)")
-
-    with env.endpoints.create(
-        "child_of_main", tenant_id=tenant_id, config_lines=config_lines
-    ) as ep:
-        # endpoint should start.
-        ep.start(safekeeper_generation=1, safekeepers=safekeeper_list)
-        ep.safe_psql("CREATE TABLE IF NOT EXISTS t(key int, value text)")
-
-    if deletetion_subject is DeletionSubject.TENANT:
-        env.storage_controller.pageserver_api().tenant_delete(tenant_id)
-    else:
-        env.storage_controller.pageserver_api().timeline_delete(tenant_id, child_timeline_id)
-
-    # ensure that there is log msgs for the third safekeeper too
-    def timeline_deleted_on_sk():
-        env.safekeepers[0].assert_log_contains(
-            f"deleting timeline {tenant_id}/{child_timeline_id} from disk"
-        )
-
-    wait_until(timeline_deleted_on_sk)
-
-
 @pytest.mark.parametrize("wrong_az", [True, False])
 def test_storage_controller_graceful_migration(neon_env_builder: NeonEnvBuilder, wrong_az: bool):
     """