From a1924e72ad5d14df37af15ffb157874549e6056f Mon Sep 17 00:00:00 2001
From: Vlad Lazar <vlad@neon.tech>
Date: Fri, 25 Jul 2025 17:41:07 +0100
Subject: [PATCH] storcon: port over hadron persistence changes

---
 Cargo.lock                                | 872 ++++++++++++++++------
 storage_controller/Cargo.toml             |   6 +-
 storage_controller/src/hadron_dns.rs      |   7 +
 storage_controller/src/hadron_queries.rs  | 848 +++++++++++++++++++++
 storage_controller/src/hadron_requests.rs |  34 +
 storage_controller/src/lib.rs             |   4 +
 storage_controller/src/persistence.rs     | 194 ++++-
 storage_controller/src/sk_node.rs         |  54 ++
 8 files changed, 1803 insertions(+), 216 deletions(-)
 create mode 100644 storage_controller/src/hadron_dns.rs
 create mode 100644 storage_controller/src/hadron_queries.rs
 create mode 100644 storage_controller/src/hadron_requests.rs
 create mode 100644 storage_controller/src/sk_node.rs

diff --git a/Cargo.lock b/Cargo.lock
index 084f867c57..74c17b8047 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -17,12 +17,6 @@ dependencies = [
  "gimli",
 ]
 
-[[package]]
-name = "adler"
-version = "1.0.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f26201604c87b1e01bd3d98f8d5d9a8fcbb815e8cedb41ffccbeb4bf593a35fe"
-
 [[package]]
 name = "adler2"
 version = "2.0.0"
@@ -145,9 +139,9 @@ dependencies = [
 
 [[package]]
 name = "anyhow"
-version = "1.0.94"
+version = "1.0.98"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c1fd03a028ef38ba2276dce7e33fcd6369c158a1bca17946c4b1b701891c1ff7"
+checksum = "e16d2d3311acee920a9eb8d33b8cbc1787ce4a264e85f964c2404b969bdcd487"
 dependencies = [
  "backtrace",
 ]
@@ -244,15 +238,24 @@ dependencies = [
 
 [[package]]
 name = "async-trait"
-version = "0.1.68"
+version = "0.1.88"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b9ccdd8f2a161be9bd5c023df56f1b2a0bd1d83872ae53b71a84a12c9bf6e842"
+checksum = "e539d3fca749fcee5236ab05e93a52867dd549cc157c8cb7f99595f3cedffdb5"
 dependencies = [
  "proc-macro2",
  "quote",
  "syn 2.0.100",
 ]
 
+[[package]]
+name = "atoi"
+version = "2.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f28d99ec8bfea296261ca1af174f24225171fea9664ba9003cbebee704810528"
+dependencies = [
+ "num-traits",
+]
+
 [[package]]
 name = "atomic-take"
 version = "1.1.0"
@@ -707,7 +710,7 @@ dependencies = [
  "http 1.3.1",
  "http-body 1.0.0",
  "http-body-util",
- "hyper 1.4.1",
+ "hyper 1.6.0",
  "hyper-util",
  "itoa",
  "matchit",
@@ -724,7 +727,7 @@ dependencies = [
  "sync_wrapper 1.0.1",
  "tokio",
  "tokio-tungstenite 0.26.1",
- "tower 0.5.2",
+ "tower",
  "tower-layer",
  "tower-service",
  "tracing",
@@ -770,7 +773,7 @@ dependencies = [
  "serde",
  "serde_html_form",
  "serde_path_to_error",
- "tower 0.5.2",
+ "tower",
  "tower-layer",
  "tower-service",
 ]
@@ -885,7 +888,7 @@ dependencies = [
  "addr2line",
  "cfg-if",
  "libc",
- "miniz_oxide 0.8.0",
+ "miniz_oxide",
  "object",
  "rustc-demangle",
  "windows-targets 0.52.6",
@@ -1348,7 +1351,7 @@ dependencies = [
  "hostname-validator",
  "http 1.3.1",
  "http-body-util",
- "hyper 1.4.1",
+ "hyper 1.6.0",
  "hyper-util",
  "indexmap 2.10.0",
  "itertools 0.10.5",
@@ -1384,7 +1387,7 @@ dependencies = [
  "tokio-stream",
  "tokio-util",
  "tonic",
- "tower 0.5.2",
+ "tower",
  "tower-http",
  "tower-otel",
  "tracing",
@@ -1547,6 +1550,21 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "crc"
+version = "3.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9710d3b3739c2e349eb44fe848ad0b7c8cb1e42bd87ee49371df2f7acaf3e675"
+dependencies = [
+ "crc-catalog",
+]
+
+[[package]]
+name = "crc-catalog"
+version = "2.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "19d374276b40fb8bbdee95aef7c7fa6b5316ec764510eb64b8dd0e2ed0d7e7f5"
+
 [[package]]
 name = "crc32c"
 version = "0.6.8"
@@ -1641,6 +1659,15 @@ dependencies = [
  "crossbeam-utils",
 ]
 
+[[package]]
+name = "crossbeam-queue"
+version = "0.3.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0f58bbc28f91df819d0aa2a2c00cd19754769c2fad90579b3592b1c9ba7a3115"
+dependencies = [
+ "crossbeam-utils",
+]
+
 [[package]]
 name = "crossbeam-utils"
 version = "0.8.19"
@@ -1978,6 +2005,12 @@ dependencies = [
  "const-random",
 ]
 
+[[package]]
+name = "dotenvy"
+version = "0.15.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1aaf95b3e5c8f23aa320147307562d361db0ae0d51242340f558153b4eb2439b"
+
 [[package]]
 name = "dsl_auto_type"
 version = "0.1.1"
@@ -2051,6 +2084,9 @@ name = "either"
 version = "1.8.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7fcaabb2fef8c910e7f4c7ce9f67a1283a1715879a7c230ca9d6d1ae31f16d91"
+dependencies = [
+ "serde",
+]
 
 [[package]]
 name = "elliptic-curve"
@@ -2117,7 +2153,7 @@ dependencies = [
  "test-log",
  "tokio",
  "tokio-util",
- "tower 0.5.2",
+ "tower",
  "tracing",
  "utils",
  "workspace_hack",
@@ -2216,12 +2252,23 @@ checksum = "5443807d6dff69373d433ab9ef5378ad8df50ca6298caf15de6e52e24aaf54d5"
 
 [[package]]
 name = "errno"
-version = "0.3.8"
+version = "0.3.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a258e46cdc063eb8519c00b9fc845fc47bcfca4130e2f08e88665ceda8474245"
+checksum = "778e2ac28f6c47af28e4907f13ffd1e1ddbd400980a9abd7c8df189bf578a5ad"
 dependencies = [
  "libc",
- "windows-sys 0.52.0",
+ "windows-sys 0.59.0",
+]
+
+[[package]]
+name = "etcetera"
+version = "0.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "136d1b5283a1ab77bd9257427ffd09d8667ced0570b6f938942bc7568ed5b943"
+dependencies = [
+ "cfg-if",
+ "home",
+ "windows-sys 0.48.0",
 ]
 
 [[package]]
@@ -2347,12 +2394,12 @@ checksum = "b3ea1ec5f8307826a5b71094dd91fc04d4ae75d5709b20ad351c7fb4815c86ec"
 
 [[package]]
 name = "flate2"
-version = "1.0.26"
+version = "1.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3b9429470923de8e8cbd4d2dc513535400b4b3fef0319fb5c4e1f520a7bef743"
+checksum = "4a3d7db9596fecd151c5f638c0ee5d5bd487b6e0ea232e5dc96d5250f6f94b1d"
 dependencies = [
  "crc32fast",
- "miniz_oxide 0.7.1",
+ "miniz_oxide",
 ]
 
 [[package]]
@@ -2361,6 +2408,27 @@ version = "1.0.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1"
 
+[[package]]
+name = "foldhash"
+version = "0.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2"
+
+[[package]]
+name = "foreign-types"
+version = "0.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f6f339eb8adc052cd2ca78910fda869aefa38d22d5cb648e6485e4d3fc06f3b1"
+dependencies = [
+ "foreign-types-shared",
+]
+
+[[package]]
+name = "foreign-types-shared"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "00b0228411908ca8685dba7fc2cdd70ec9990a6e753e89b6ac91a84c40fbaf4b"
+
 [[package]]
 name = "form_urlencoded"
 version = "1.2.1"
@@ -2381,7 +2449,7 @@ dependencies = [
  "futures-core",
  "futures-sink",
  "http-body-util",
- "hyper 1.4.1",
+ "hyper 1.6.0",
  "hyper-util",
  "pin-project",
  "rand 0.8.5",
@@ -2442,6 +2510,17 @@ dependencies = [
  "futures-util",
 ]
 
+[[package]]
+name = "futures-intrusive"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1d930c203dd0b6ff06e0201a4a2fe9149b43c684fd4420555b26d21b1a02956f"
+dependencies = [
+ "futures-core",
+ "lock_api",
+ "parking_lot 0.12.1",
+]
+
 [[package]]
 name = "futures-io"
 version = "0.3.31"
@@ -2572,9 +2651,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "26145e563e54f2cadc477553f1ec5ee650b00862f0a58bcd12cbdc5f0ea2d2f4"
 dependencies = [
  "cfg-if",
+ "js-sys",
  "libc",
  "r-efi",
  "wasi 0.14.2+wasi-0.2.4",
+ "wasm-bindgen",
 ]
 
 [[package]]
@@ -2741,6 +2822,11 @@ name = "hashbrown"
 version = "0.15.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bf151400ff0baff5465007dd2f3e717f3fe502074ca563069ce3a6629d07b289"
+dependencies = [
+ "allocator-api2",
+ "equivalent",
+ "foldhash",
+]
 
 [[package]]
 name = "hashlink"
@@ -2751,6 +2837,15 @@ dependencies = [
  "hashbrown 0.14.5",
 ]
 
+[[package]]
+name = "hashlink"
+version = "0.10.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7382cf6263419f2d8df38c55d7da83da5c18aef87fc7a7fc1fb1e344edfe14c1"
+dependencies = [
+ "hashbrown 0.15.2",
+]
+
 [[package]]
 name = "hdrhistogram"
 version = "7.5.4"
@@ -2831,6 +2926,15 @@ dependencies = [
  "never-say-never",
 ]
 
+[[package]]
+name = "hkdf"
+version = "0.12.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7b5f8eb2ad728638ea2c7d47a21db23b7b58a72ed6a38256b8a1849f15fbbdf7"
+dependencies = [
+ "hmac",
+]
+
 [[package]]
 name = "hmac"
 version = "0.12.1"
@@ -2840,6 +2944,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"
@@ -2972,9 +3085,9 @@ dependencies = [
 
 [[package]]
 name = "httparse"
-version = "1.8.0"
+version = "1.10.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d897f394bad6a705d5f4104762e116a75639e470d80901eed05a860a95cb1904"
+checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87"
 
 [[package]]
 name = "httpdate"
@@ -3015,7 +3128,7 @@ dependencies = [
  "httpdate",
  "itoa",
  "pin-project-lite",
- "socket2",
+ "socket2 0.5.5",
  "tokio",
  "tower-service",
  "tracing",
@@ -3024,9 +3137,9 @@ dependencies = [
 
 [[package]]
 name = "hyper"
-version = "1.4.1"
+version = "1.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "50dfd22e0e76d0f662d429a5f80fcaf3855009297eab6a0a9f8543834744ba05"
+checksum = "cc2b571658e38e0c01b1fdca3bbbe93c00d3d71693ff2770043f8c29bc7d6f80"
 dependencies = [
  "bytes",
  "futures-channel",
@@ -3060,19 +3173,20 @@ dependencies = [
 
 [[package]]
 name = "hyper-rustls"
-version = "0.26.0"
+version = "0.27.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a0bea761b46ae2b24eb4aef630d8d1c398157b6fc29e6350ecf090a0b70c952c"
+checksum = "e3c93eb611681b207e1fe55d5a71ecf91572ec8a6705cdb6857f7d8d5242cf58"
 dependencies = [
- "futures-util",
  "http 1.3.1",
- "hyper 1.4.1",
+ "hyper 1.6.0",
  "hyper-util",
- "rustls 0.22.4",
+ "rustls 0.23.29",
+ "rustls-native-certs 0.8.0",
  "rustls-pki-types",
  "tokio",
- "tokio-rustls 0.25.0",
+ "tokio-rustls 0.26.2",
  "tower-service",
+ "webpki-roots 1.0.2",
 ]
 
 [[package]]
@@ -3081,7 +3195,7 @@ version = "0.5.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3203a961e5c83b6f5498933e78b6b263e208c197b63e9c6c53cc82ffd3f63793"
 dependencies = [
- "hyper 1.4.1",
+ "hyper 1.6.0",
  "hyper-util",
  "pin-project-lite",
  "tokio",
@@ -3089,21 +3203,41 @@ dependencies = [
 ]
 
 [[package]]
-name = "hyper-util"
-version = "0.1.7"
+name = "hyper-tls"
+version = "0.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cde7055719c54e36e95e8719f95883f22072a48ede39db7fc17a4e1d5281e9b9"
+checksum = "70206fc6890eaca9fde8a0bf71caa2ddfc9fe045ac9e5c70df101a7dbde866e0"
 dependencies = [
+ "bytes",
+ "http-body-util",
+ "hyper 1.6.0",
+ "hyper-util",
+ "native-tls",
+ "tokio",
+ "tokio-native-tls",
+ "tower-service",
+]
+
+[[package]]
+name = "hyper-util"
+version = "0.1.16"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8d9b05277c7e8da2c93a568989bb6207bef0112e8d17df7a6eda4a3cf143bc5e"
+dependencies = [
+ "base64 0.22.1",
  "bytes",
  "futures-channel",
+ "futures-core",
  "futures-util",
  "http 1.3.1",
  "http-body 1.0.0",
- "hyper 1.4.1",
+ "hyper 1.6.0",
+ "ipnet",
+ "libc",
+ "percent-encoding",
  "pin-project-lite",
- "socket2",
+ "socket2 0.6.0",
  "tokio",
- "tower 0.4.13",
  "tower-service",
  "tracing",
 ]
@@ -3398,12 +3532,33 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "io-uring"
+version = "0.7.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d93587f37623a1a17d94ef2bc9ada592f5465fe7732084ab7beefabe5c77c0c4"
+dependencies = [
+ "bitflags 2.8.0",
+ "cfg-if",
+ "libc",
+]
+
 [[package]]
 name = "ipnet"
 version = "2.10.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ddc24109865250148c2e0f3d25d4f0f479571723792d3802153c60922a4fb708"
 
+[[package]]
+name = "iri-string"
+version = "0.7.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dbc5ebe9c3a1a7a5127f920a418f7585e9e758e911d0466ed004f393b0e380b2"
+dependencies = [
+ "memchr",
+ "serde",
+]
+
 [[package]]
 name = "is-terminal"
 version = "0.4.12"
@@ -3533,10 +3688,11 @@ dependencies = [
 
 [[package]]
 name = "js-sys"
-version = "0.3.69"
+version = "0.3.77"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "29c15563dc2726973df627357ce0c9ddddbea194836909d655df6a75d2cf296d"
+checksum = "1cfaf33c695fc6e08064efbc1f72ec937429614f25eef83af942d0e227c3a28f"
 dependencies = [
+ "once_cell",
  "wasm-bindgen",
 ]
 
@@ -3648,6 +3804,12 @@ version = "0.6.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f0b5399f6804fbab912acbd8878ed3532d506b7c951b8f9f164ef90fef39e3f4"
 
+[[package]]
+name = "linux-raw-sys"
+version = "0.9.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cd945864f07fe9f5371a27ad7b52a172b4b499999f1d97574c9fa68373937e12"
+
 [[package]]
 name = "litemap"
 version = "0.7.4"
@@ -3692,6 +3854,12 @@ dependencies = [
  "hashbrown 0.14.5",
 ]
 
+[[package]]
+name = "lru-slab"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154"
+
 [[package]]
 name = "mappings"
 version = "0.7.0"
@@ -3860,18 +4028,9 @@ checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a"
 
 [[package]]
 name = "miniz_oxide"
-version = "0.7.1"
+version = "0.8.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e7810e0be55b428ada41041c41f32c9f1a42817901b4ccf45fa3d4b6561e74c7"
-dependencies = [
- "adler",
-]
-
-[[package]]
-name = "miniz_oxide"
-version = "0.8.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e2d80299ef12ff69b16a84bb182e3b9df68b5a91574d3d4fa6e41b65deec4df1"
+checksum = "1fa76a2c86f704bdb222d66965fb3d63269ce38518b83cb0575fca855ebb6316"
 dependencies = [
  "adler2",
 ]
@@ -3924,6 +4083,23 @@ version = "0.8.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e5ce46fe64a9d73be07dcbe690a38ce1b293be448fd8ce1e6c1b8062c9f72c6a"
 
+[[package]]
+name = "native-tls"
+version = "0.2.14"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "87de3442987e9dbec73158d5c715e7ad9072fda936bb03d19d7fa10e00520f0e"
+dependencies = [
+ "libc",
+ "log",
+ "openssl",
+ "openssl-probe",
+ "openssl-sys",
+ "schannel",
+ "security-framework",
+ "security-framework-sys",
+ "tempfile",
+]
+
 [[package]]
 name = "neon-shmem"
 version = "0.1.0"
@@ -4201,12 +4377,50 @@ version = "11.1.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0ab1bc2a289d34bd04a330323ac98a1b4bc82c9d9fcb1e66b63caa84da26b575"
 
+[[package]]
+name = "openssl"
+version = "0.10.73"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8505734d46c8ab1e19a1dce3aef597ad87dcb4c37e7188231769bd6bd51cebf8"
+dependencies = [
+ "bitflags 2.8.0",
+ "cfg-if",
+ "foreign-types",
+ "libc",
+ "once_cell",
+ "openssl-macros",
+ "openssl-sys",
+]
+
+[[package]]
+name = "openssl-macros"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a948666b637a0f465e8564c73e89d4dde00d72d4d473cc972f390fc3dcee7d9c"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.100",
+]
+
 [[package]]
 name = "openssl-probe"
 version = "0.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ff011a302c396a5197692431fc1948019154afc178baf7d8e37367442a4601cf"
 
+[[package]]
+name = "openssl-sys"
+version = "0.9.109"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "90096e2e47630d78b7d1c20952dc621f957103f8bc2c8359ec81290d75238571"
+dependencies = [
+ "cc",
+ "libc",
+ "pkg-config",
+ "vcpkg",
+]
+
 [[package]]
 name = "opentelemetry"
 version = "0.30.0"
@@ -4217,7 +4431,7 @@ dependencies = [
  "futures-sink",
  "js-sys",
  "pin-project-lite",
- "thiserror 2.0.11",
+ "thiserror 2.0.12",
  "tracing",
 ]
 
@@ -4247,7 +4461,7 @@ dependencies = [
  "opentelemetry_sdk",
  "prost 0.13.5",
  "reqwest",
- "thiserror 2.0.11",
+ "thiserror 2.0.12",
 ]
 
 [[package]]
@@ -4281,7 +4495,7 @@ dependencies = [
  "percent-encoding",
  "rand 0.9.1",
  "serde_json",
- "thiserror 2.0.11",
+ "thiserror 2.0.12",
  "tokio",
  "tokio-stream",
 ]
@@ -4466,7 +4680,7 @@ dependencies = [
  "enumset",
  "fail",
  "futures",
- "hashlink",
+ "hashlink 0.9.1",
  "hex",
  "hex-literal",
  "http 1.3.1",
@@ -4533,7 +4747,7 @@ dependencies = [
  "toml_edit",
  "tonic",
  "tonic-reflection",
- "tower 0.5.2",
+ "tower",
  "tracing",
  "tracing-utils",
  "twox-hash",
@@ -5122,6 +5336,65 @@ dependencies = [
  "workspace_hack",
 ]
 
+[[package]]
+name = "postgresql_archive"
+version = "0.19.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b2b10a6a63e5f4ccf5d99d7d5db0829aec30a5e84b5c80883f6561c77a7521a6"
+dependencies = [
+ "async-trait",
+ "flate2",
+ "futures-util",
+ "hex",
+ "num-format",
+ "regex-lite",
+ "reqwest",
+ "reqwest-middleware",
+ "reqwest-retry",
+ "reqwest-tracing",
+ "semver",
+ "serde",
+ "serde_json",
+ "sha2",
+ "tar",
+ "target-triple",
+ "tempfile",
+ "thiserror 2.0.12",
+ "tracing",
+ "url",
+]
+
+[[package]]
+name = "postgresql_commands"
+version = "0.19.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9cc8653c31df5ffcd3bc114641dfae5390404d6e22bc21fd7473760f22cdc20b"
+dependencies = [
+ "thiserror 2.0.12",
+ "tokio",
+ "tracing",
+]
+
+[[package]]
+name = "postgresql_embedded"
+version = "0.19.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4386829807c8dfbc190f8e53b91d2080f0f604bd4b9f0831281ccfbc06998123"
+dependencies = [
+ "anyhow",
+ "postgresql_archive",
+ "postgresql_commands",
+ "rand 0.9.1",
+ "semver",
+ "sqlx",
+ "target-triple",
+ "tempfile",
+ "thiserror 2.0.12",
+ "tokio",
+ "tracing",
+ "url",
+]
+
 [[package]]
 name = "posthog_client_lite"
 version = "0.1.0"
@@ -5261,7 +5534,7 @@ dependencies = [
  "hex",
  "lazy_static",
  "procfs-core",
- "rustix",
+ "rustix 0.38.41",
 ]
 
 [[package]]
@@ -5431,7 +5704,7 @@ dependencies = [
  "futures",
  "gettid",
  "hashbrown 0.14.5",
- "hashlink",
+ "hashlink 0.9.1",
  "hex",
  "hmac",
  "hostname",
@@ -5441,7 +5714,7 @@ dependencies = [
  "humantime",
  "humantime-serde",
  "hyper 0.14.30",
- "hyper 1.4.1",
+ "hyper 1.6.0",
  "hyper-util",
  "indexmap 2.10.0",
  "ipnet",
@@ -5491,7 +5764,7 @@ dependencies = [
  "signature 2.2.0",
  "smallvec",
  "smol_str",
- "socket2",
+ "socket2 0.5.5",
  "strum_macros",
  "subtle",
  "subzero-core",
@@ -5565,6 +5838,61 @@ dependencies = [
  "memchr",
 ]
 
+[[package]]
+name = "quinn"
+version = "0.11.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "626214629cda6781b6dc1d316ba307189c85ba657213ce642d9c77670f8202c8"
+dependencies = [
+ "bytes",
+ "cfg_aliases",
+ "pin-project-lite",
+ "quinn-proto",
+ "quinn-udp",
+ "rustc-hash 2.1.1",
+ "rustls 0.23.29",
+ "socket2 0.5.5",
+ "thiserror 2.0.12",
+ "tokio",
+ "tracing",
+ "web-time",
+]
+
+[[package]]
+name = "quinn-proto"
+version = "0.11.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "49df843a9161c85bb8aae55f101bc0bac8bcafd637a620d9122fd7e0b2f7422e"
+dependencies = [
+ "bytes",
+ "getrandom 0.3.3",
+ "lru-slab",
+ "rand 0.9.1",
+ "ring",
+ "rustc-hash 2.1.1",
+ "rustls 0.23.29",
+ "rustls-pki-types",
+ "slab",
+ "thiserror 2.0.12",
+ "tinyvec",
+ "tracing",
+ "web-time",
+]
+
+[[package]]
+name = "quinn-udp"
+version = "0.5.13"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fcebb1209ee276352ef14ff8732e24cc2b02bbac986cd74a4c81bcb2f9881970"
+dependencies = [
+ "cfg_aliases",
+ "libc",
+ "once_cell",
+ "socket2 0.5.5",
+ "tracing",
+ "windows-sys 0.59.0",
+]
+
 [[package]]
 name = "quote"
 version = "1.0.39"
@@ -5764,7 +6092,7 @@ dependencies = [
  "rustls-native-certs 0.8.0",
  "ryu",
  "sha1_smol",
- "socket2",
+ "socket2 0.5.5",
  "tokio",
  "tokio-rustls 0.26.2",
  "tokio-util",
@@ -5832,9 +6160,9 @@ dependencies = [
 
 [[package]]
 name = "regex-lite"
-version = "0.1.5"
+version = "0.1.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "30b661b2f27137bdbc16f00eda72866a92bb28af1753ffbd56744fb6e2e9cd8e"
+checksum = "53a49587ad06b26609c52e423de037e7f57f20d53535d66e08c695f347df952a"
 
 [[package]]
 name = "regex-syntax"
@@ -5879,7 +6207,7 @@ dependencies = [
  "http-body-util",
  "http-types",
  "humantime-serde",
- "hyper 1.4.1",
+ "hyper 1.6.0",
  "itertools 0.10.5",
  "metrics",
  "once_cell",
@@ -5907,9 +6235,9 @@ checksum = "e3a8614ee435691de62bcffcf4a66d91b3594bf1428a5722e79103249a095690"
 
 [[package]]
 name = "reqwest"
-version = "0.12.4"
+version = "0.12.22"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "566cafdd92868e0939d3fb961bd0dc25fcfaaed179291093b3d43e6b3150ea10"
+checksum = "cbc931937e6ca3a06e3b6c0aa7841849b160a90351d6ab467a8b9b9959767531"
 dependencies = [
  "base64 0.22.1",
  "bytes",
@@ -5919,42 +6247,43 @@ dependencies = [
  "http 1.3.1",
  "http-body 1.0.0",
  "http-body-util",
- "hyper 1.4.1",
- "hyper-rustls 0.26.0",
+ "hyper 1.6.0",
+ "hyper-rustls 0.27.7",
+ "hyper-tls",
  "hyper-util",
- "ipnet",
  "js-sys",
  "log",
- "mime",
- "once_cell",
+ "native-tls",
  "percent-encoding",
  "pin-project-lite",
- "rustls 0.22.4",
- "rustls-native-certs 0.7.0",
- "rustls-pemfile 2.1.1",
+ "quinn",
+ "rustls 0.23.29",
+ "rustls-native-certs 0.8.0",
  "rustls-pki-types",
  "serde",
  "serde_json",
  "serde_urlencoded",
- "sync_wrapper 0.1.2",
+ "sync_wrapper 1.0.1",
  "tokio",
- "tokio-rustls 0.25.0",
+ "tokio-native-tls",
+ "tokio-rustls 0.26.2",
  "tokio-util",
+ "tower",
+ "tower-http",
  "tower-service",
  "url",
  "wasm-bindgen",
  "wasm-bindgen-futures",
  "wasm-streams",
  "web-sys",
- "webpki-roots",
- "winreg",
+ "webpki-roots 1.0.2",
 ]
 
 [[package]]
 name = "reqwest-middleware"
-version = "0.4.0"
+version = "0.4.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d1ccd3b55e711f91a9885a2fa6fbbb2e39db1776420b062efc058c6410f7e5e3"
+checksum = "57f17d28a6e6acfe1733fe24bcd30774d13bffa4b8a22535b4c8c98423088d4e"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -5976,7 +6305,7 @@ dependencies = [
  "futures",
  "getrandom 0.2.11",
  "http 1.3.1",
- "hyper 1.4.1",
+ "hyper 1.6.0",
  "parking_lot 0.11.2",
  "reqwest",
  "reqwest-middleware",
@@ -6188,6 +6517,19 @@ dependencies = [
  "windows-sys 0.52.0",
 ]
 
+[[package]]
+name = "rustix"
+version = "1.0.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "11181fbabf243db407ef8df94a6ce0b2f9a733bd8be4ad02b4eda9602296cac8"
+dependencies = [
+ "bitflags 2.8.0",
+ "errno",
+ "libc",
+ "linux-raw-sys 0.9.4",
+ "windows-sys 0.59.0",
+]
+
 [[package]]
 name = "rustls"
 version = "0.21.12"
@@ -6200,20 +6542,6 @@ dependencies = [
  "sct",
 ]
 
-[[package]]
-name = "rustls"
-version = "0.22.4"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bf4ef73721ac7bcd79b2b315da7779d8fc09718c6b3d2d1b2d94850eb8c18432"
-dependencies = [
- "log",
- "ring",
- "rustls-pki-types",
- "rustls-webpki 0.102.8",
- "subtle",
- "zeroize",
-]
-
 [[package]]
 name = "rustls"
 version = "0.23.29"
@@ -6241,19 +6569,6 @@ dependencies = [
  "security-framework",
 ]
 
-[[package]]
-name = "rustls-native-certs"
-version = "0.7.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8f1fb85efa936c42c6d5fc28d2629bb51e4b2f4b8a5211e297d599cc5a093792"
-dependencies = [
- "openssl-probe",
- "rustls-pemfile 2.1.1",
- "rustls-pki-types",
- "schannel",
- "security-framework",
-]
-
 [[package]]
 name = "rustls-native-certs"
 version = "0.8.0"
@@ -6292,6 +6607,7 @@ version = "1.12.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "229a4a4c221013e7e1f1a043678c5cc39fe5171437c88fb47151a21e6f5b5c79"
 dependencies = [
+ "web-time",
  "zeroize",
 ]
 
@@ -6305,17 +6621,6 @@ dependencies = [
  "untrusted",
 ]
 
-[[package]]
-name = "rustls-webpki"
-version = "0.102.8"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "64ca1bc8749bd4cf37b5ce386cc146580777b4e8572c7b97baf22c83f444bee9"
-dependencies = [
- "ring",
- "rustls-pki-types",
- "untrusted",
-]
-
 [[package]]
 name = "rustls-webpki"
 version = "0.103.4"
@@ -6559,9 +6864,9 @@ dependencies = [
 
 [[package]]
 name = "semver"
-version = "1.0.17"
+version = "1.0.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bebd363326d05ec3e2f532ab7660680f3b02130d780c299bca73469d521bc0ed"
+checksum = "56e6fa9c48d24d85fb3de5ad847117517440f6beceb7798af16b4a87d616b8d0"
 
 [[package]]
 name = "send-future"
@@ -6585,7 +6890,7 @@ dependencies = [
  "sentry-tracing",
  "tokio",
  "ureq",
- "webpki-roots",
+ "webpki-roots 0.26.1",
 ]
 
 [[package]]
@@ -6674,9 +6979,9 @@ checksum = "a3f0bf26fd526d2a95683cd0f87bf103b8539e2ca1ef48ce002d67aad59aa0b4"
 
 [[package]]
 name = "serde"
-version = "1.0.215"
+version = "1.0.219"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6513c1ad0b11a9376da888e3e0baa0077f1aed55c17f50e7b2397136129fb88f"
+checksum = "5f0e2c6ed6606019b4e29e69dbaba95b11854410e5347d525002456dbbb786b6"
 dependencies = [
  "serde_derive",
 ]
@@ -6693,9 +6998,9 @@ dependencies = [
 
 [[package]]
 name = "serde_derive"
-version = "1.0.215"
+version = "1.0.219"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ad1e866f866923f252f05c889987993144fb74e722403468a4ebd70c3cd756c0"
+checksum = "5b0276cf7f2c73365f7157c8123c21cd9a50fbbd844757af28ca1f5925fc2a00"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -6717,9 +7022,9 @@ dependencies = [
 
 [[package]]
 name = "serde_json"
-version = "1.0.125"
+version = "1.0.141"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "83c8e735a073ccf5be70aa8066aa984eaf2fa000db6c8d0100ae605b366d31ed"
+checksum = "30b9eff21ebe718216c6ec64e1d9ac57087aad11efc64e32002bce4a0d4c03d3"
 dependencies = [
  "itoa",
  "memchr",
@@ -6932,18 +7237,18 @@ checksum = "7bd3e3206899af3f8b12af284fafc038cc1dc2b41d1b89dd17297221c5d225de"
 
 [[package]]
 name = "slab"
-version = "0.4.8"
+version = "0.4.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6528351c9bc8ab22353f9d776db39a20288e8d6c37ef8cfe3317cf875eecfc2d"
-dependencies = [
- "autocfg",
-]
+checksum = "04dc19736151f35336d325007ac991178d504a119863a2fcb3758cdb5e52c50d"
 
 [[package]]
 name = "smallvec"
 version = "1.13.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3c5e1a9a646d36c3599cd173a41282daf47c44583ad367b8e6837255952e5c67"
+dependencies = [
+ "serde",
+]
 
 [[package]]
 name = "smol_str"
@@ -6964,6 +7269,16 @@ dependencies = [
  "windows-sys 0.48.0",
 ]
 
+[[package]]
+name = "socket2"
+version = "0.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "233504af464074f9d066d7b5416c5f9b894a5862a6506e306f7b816cdd6f1807"
+dependencies = [
+ "libc",
+ "windows-sys 0.59.0",
+]
+
 [[package]]
 name = "spin"
 version = "0.9.8"
@@ -6999,6 +7314,125 @@ dependencies = [
  "der 0.7.8",
 ]
 
+[[package]]
+name = "sqlx"
+version = "0.8.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1fefb893899429669dcdd979aff487bd78f4064e5e7907e4269081e0ef7d97dc"
+dependencies = [
+ "sqlx-core",
+ "sqlx-macros",
+ "sqlx-postgres",
+]
+
+[[package]]
+name = "sqlx-core"
+version = "0.8.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ee6798b1838b6a0f69c007c133b8df5866302197e404e8b6ee8ed3e3a5e68dc6"
+dependencies = [
+ "base64 0.22.1",
+ "bytes",
+ "crc",
+ "crossbeam-queue",
+ "either",
+ "event-listener 5.4.0",
+ "futures-core",
+ "futures-intrusive",
+ "futures-io",
+ "futures-util",
+ "hashbrown 0.15.2",
+ "hashlink 0.10.0",
+ "indexmap 2.10.0",
+ "log",
+ "memchr",
+ "native-tls",
+ "once_cell",
+ "percent-encoding",
+ "serde",
+ "serde_json",
+ "sha2",
+ "smallvec",
+ "thiserror 2.0.12",
+ "tokio",
+ "tokio-stream",
+ "tracing",
+ "url",
+]
+
+[[package]]
+name = "sqlx-macros"
+version = "0.8.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a2d452988ccaacfbf5e0bdbc348fb91d7c8af5bee192173ac3636b5fb6e6715d"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "sqlx-core",
+ "sqlx-macros-core",
+ "syn 2.0.100",
+]
+
+[[package]]
+name = "sqlx-macros-core"
+version = "0.8.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "19a9c1841124ac5a61741f96e1d9e2ec77424bf323962dd894bdb93f37d5219b"
+dependencies = [
+ "dotenvy",
+ "either",
+ "heck 0.5.0",
+ "hex",
+ "once_cell",
+ "proc-macro2",
+ "quote",
+ "serde",
+ "serde_json",
+ "sha2",
+ "sqlx-core",
+ "sqlx-postgres",
+ "syn 2.0.100",
+ "tokio",
+ "url",
+]
+
+[[package]]
+name = "sqlx-postgres"
+version = "0.8.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "db58fcd5a53cf07c184b154801ff91347e4c30d17a3562a635ff028ad5deda46"
+dependencies = [
+ "atoi",
+ "base64 0.22.1",
+ "bitflags 2.8.0",
+ "byteorder",
+ "crc",
+ "dotenvy",
+ "etcetera",
+ "futures-channel",
+ "futures-core",
+ "futures-util",
+ "hex",
+ "hkdf",
+ "hmac",
+ "home",
+ "itoa",
+ "log",
+ "md-5",
+ "memchr",
+ "once_cell",
+ "rand 0.8.5",
+ "serde",
+ "serde_json",
+ "sha2",
+ "smallvec",
+ "sqlx-core",
+ "stringprep",
+ "thiserror 2.0.12",
+ "tracing",
+ "whoami",
+]
+
 [[package]]
 name = "stable_deref_trait"
 version = "1.2.0"
@@ -7027,7 +7461,7 @@ dependencies = [
  "http-body-util",
  "http-utils",
  "humantime",
- "hyper 1.4.1",
+ "hyper 1.6.0",
  "hyper-util",
  "metrics",
  "once_cell",
@@ -7076,6 +7510,8 @@ dependencies = [
  "pageserver_api",
  "pageserver_client",
  "postgres_connection",
+ "postgresql_archive",
+ "postgresql_embedded",
  "posthog_client_lite",
  "rand 0.9.1",
  "regex",
@@ -7295,6 +7731,9 @@ name = "sync_wrapper"
 version = "1.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a7065abeca94b6a8a577f9bd45aa0867a2238b74e8eb67cf10d492bc39351394"
+dependencies = [
+ "futures-core",
+]
 
 [[package]]
 name = "synstructure"
@@ -7330,9 +7769,9 @@ checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417"
 
 [[package]]
 name = "tar"
-version = "0.4.40"
+version = "0.4.44"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b16afcea1f22891c49a00c751c7b63b2233284064f11a200fc624137c51e2ddb"
+checksum = "1d863878d212c87a19c1a610eb53bb01fe12951c0501cf5a0d65f724914a667a"
 dependencies = [
  "filetime",
  "libc",
@@ -7340,15 +7779,21 @@ dependencies = [
 ]
 
 [[package]]
-name = "tempfile"
-version = "3.14.0"
+name = "target-triple"
+version = "0.1.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "28cce251fcbc87fac86a866eeb0d6c2d536fc16d06f184bb61aeae11aa4cee0c"
+checksum = "1ac9aa371f599d22256307c24a9d748c041e548cbf599f35d890f9d365361790"
+
+[[package]]
+name = "tempfile"
+version = "3.20.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e8a64e3985349f2441a1a9ef0b853f869006c3855f2cda6862a94d26ebb9d6a1"
 dependencies = [
- "cfg-if",
  "fastrand 2.2.0",
+ "getrandom 0.3.3",
  "once_cell",
- "rustix",
+ "rustix 1.0.8",
  "windows-sys 0.59.0",
 ]
 
@@ -7415,11 +7860,11 @@ dependencies = [
 
 [[package]]
 name = "thiserror"
-version = "2.0.11"
+version = "2.0.12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d452f284b73e6d76dd36758a0c8684b1d5be31f92b89d07fd5822175732206fc"
+checksum = "567b8a2dae586314f7be2a752ec7474332959c6460e02bde30d702a66d488708"
 dependencies = [
- "thiserror-impl 2.0.11",
+ "thiserror-impl 2.0.12",
 ]
 
 [[package]]
@@ -7435,9 +7880,9 @@ dependencies = [
 
 [[package]]
 name = "thiserror-impl"
-version = "2.0.11"
+version = "2.0.12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "26afc1baea8a989337eeb52b6e72a039780ce45c3edfcc9c5b9d112feeb173c2"
+checksum = "7f7cf42b4507d8ea322120659672cf1b9dbb93f8f2d4ecfd6e51350ff5b17a1d"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -7595,18 +8040,20 @@ dependencies = [
 
 [[package]]
 name = "tokio"
-version = "1.43.1"
+version = "1.46.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "492a604e2fd7f814268a378409e6c92b5525d747d10db9a229723f55a417958c"
+checksum = "0cc3a2344dafbe23a245241fe8b09735b521110d30fcefbbd5feb1797ca35d17"
 dependencies = [
  "backtrace",
  "bytes",
+ "io-uring 0.7.9",
  "libc",
  "mio 1.0.3",
  "parking_lot 0.12.1",
  "pin-project-lite",
  "signal-hook-registry",
- "socket2",
+ "slab",
+ "socket2 0.5.5",
  "tokio-macros",
  "windows-sys 0.52.0",
 ]
@@ -7648,6 +8095,16 @@ dependencies = [
  "syn 2.0.100",
 ]
 
+[[package]]
+name = "tokio-native-tls"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bbae76ab933c85776efabc971569dd6119c580d8f5d448769dec1764bf796ef2"
+dependencies = [
+ "native-tls",
+ "tokio",
+]
+
 [[package]]
 name = "tokio-postgres"
 version = "0.7.10"
@@ -7667,7 +8124,7 @@ dependencies = [
  "postgres-protocol",
  "postgres-types",
  "rand 0.8.5",
- "socket2",
+ "socket2 0.5.5",
  "tokio",
  "tokio-util",
  "whoami",
@@ -7714,17 +8171,6 @@ dependencies = [
  "tokio",
 ]
 
-[[package]]
-name = "tokio-rustls"
-version = "0.25.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "775e0c0f0adb3a2f22a00c4745d728b479985fc15ee7ca6a2608388c5569860f"
-dependencies = [
- "rustls 0.22.4",
- "rustls-pki-types",
- "tokio",
-]
-
 [[package]]
 name = "tokio-rustls"
 version = "0.26.2"
@@ -7852,18 +8298,18 @@ dependencies = [
  "http 1.3.1",
  "http-body 1.0.0",
  "http-body-util",
- "hyper 1.4.1",
+ "hyper 1.6.0",
  "hyper-timeout",
  "hyper-util",
  "percent-encoding",
  "pin-project",
  "prost 0.13.5",
  "rustls-native-certs 0.8.0",
- "socket2",
+ "socket2 0.5.5",
  "tokio",
  "tokio-rustls 0.26.2",
  "tokio-stream",
- "tower 0.5.2",
+ "tower",
  "tower-layer",
  "tower-service",
  "tracing",
@@ -7897,21 +8343,6 @@ dependencies = [
  "tonic",
 ]
 
-[[package]]
-name = "tower"
-version = "0.4.13"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b8fa9be0de6cf49e536ce1851f987bd21a43b771b09473c3549a6c853db37c1c"
-dependencies = [
- "futures-core",
- "futures-util",
- "pin-project",
- "pin-project-lite",
- "tokio",
- "tower-layer",
- "tower-service",
-]
-
 [[package]]
 name = "tower"
 version = "0.5.2"
@@ -7933,17 +8364,20 @@ dependencies = [
 
 [[package]]
 name = "tower-http"
-version = "0.6.2"
+version = "0.6.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "403fa3b783d4b626a8ad51d766ab03cb6d2dbfc46b1c5d4448395e6628dc9697"
+checksum = "adc82fd73de2a9722ac5da747f12383d2bfdb93591ee6c58486e0097890f05f2"
 dependencies = [
  "base64 0.22.1",
  "bitflags 2.8.0",
  "bytes",
+ "futures-util",
  "http 1.3.1",
  "http-body 1.0.0",
+ "iri-string",
  "mime",
  "pin-project-lite",
+ "tower",
  "tower-layer",
  "tower-service",
  "tracing",
@@ -8171,7 +8605,7 @@ dependencies = [
  "log",
  "rand 0.8.5",
  "sha1",
- "thiserror 2.0.11",
+ "thiserror 2.0.12",
  "utf-8",
 ]
 
@@ -8261,7 +8695,7 @@ dependencies = [
  "rustls 0.23.29",
  "rustls-pki-types",
  "url",
- "webpki-roots",
+ "webpki-roots 0.26.1",
 ]
 
 [[package]]
@@ -8270,7 +8704,7 @@ version = "0.1.0"
 source = "git+https://github.com/neondatabase/tokio-epoll-uring.git?branch=main#781989bb540a1408b0b93daa1e9d1fa452195497"
 dependencies = [
  "bytes",
- "io-uring",
+ "io-uring 0.6.2",
  "libc",
  "linux-raw-sys 0.6.4",
 ]
@@ -8388,6 +8822,12 @@ version = "0.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "830b7e5d4d90034032940e4ace0d9a9a057e7a45cd94e6c007832e39edb82f6d"
 
+[[package]]
+name = "vcpkg"
+version = "0.2.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426"
+
 [[package]]
 name = "version_check"
 version = "0.9.4"
@@ -8528,23 +8968,24 @@ checksum = "b8dad83b4f25e74f184f64c43b150b91efe7647395b42289f38e50566d82855b"
 
 [[package]]
 name = "wasm-bindgen"
-version = "0.2.92"
+version = "0.2.100"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4be2531df63900aeb2bca0daaaddec08491ee64ceecbee5076636a3b026795a8"
+checksum = "1edc8929d7499fc4e8f0be2262a241556cfc54a0bea223790e71446f2aab1ef5"
 dependencies = [
  "cfg-if",
+ "once_cell",
+ "rustversion",
  "wasm-bindgen-macro",
 ]
 
 [[package]]
 name = "wasm-bindgen-backend"
-version = "0.2.92"
+version = "0.2.100"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "614d787b966d3989fa7bb98a654e369c762374fd3213d212cfc0251257e747da"
+checksum = "2f0a0651a5c2bc21487bde11ee802ccaf4c51935d0d3d42a6101f98161700bc6"
 dependencies = [
  "bumpalo",
  "log",
- "once_cell",
  "proc-macro2",
  "quote",
  "syn 2.0.100",
@@ -8565,9 +9006,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-macro"
-version = "0.2.92"
+version = "0.2.100"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a1f8823de937b71b9460c0c34e25f3da88250760bec0ebac694b49997550d726"
+checksum = "7fe63fc6d09ed3792bd0897b314f53de8e16568c2b3f7982f468c0bf9bd0b407"
 dependencies = [
  "quote",
  "wasm-bindgen-macro-support",
@@ -8575,9 +9016,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-macro-support"
-version = "0.2.92"
+version = "0.2.100"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e94f17b526d0a461a191c78ea52bbce64071ed5c04c9ffe424dcb38f74171bb7"
+checksum = "8ae87ea40c9f689fc23f209965b6fb8a99ad69aeeb0231408be24920604395de"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -8588,9 +9029,12 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-shared"
-version = "0.2.92"
+version = "0.2.100"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "af190c94f2773fdb3729c55b007a722abb5384da03bc0986df4c289bf5567e96"
+checksum = "1a05d73b933a847d6cccdda8f838a22ff101ad9bf93e33684f39c1f5f0eece3d"
+dependencies = [
+ "unicode-ident",
+]
 
 [[package]]
 name = "wasm-streams"
@@ -8649,6 +9093,15 @@ dependencies = [
  "rustls-pki-types",
 ]
 
+[[package]]
+name = "webpki-roots"
+version = "1.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7e8983c3ab33d6fb807cfcdad2491c4ea8cbc8ed839181c7dfd9c67c83e261b2"
+dependencies = [
+ "rustls-pki-types",
+]
+
 [[package]]
 name = "whoami"
 version = "1.5.1"
@@ -8978,16 +9431,6 @@ dependencies = [
  "memchr",
 ]
 
-[[package]]
-name = "winreg"
-version = "0.52.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a277a57398d4bfa075df44f501a17cfdf8542d224f0d36095a2adc7aee4ef0a5"
-dependencies = [
- "cfg-if",
- "windows-sys 0.48.0",
-]
-
 [[package]]
 name = "wit-bindgen-rt"
 version = "0.39.0"
@@ -9041,7 +9484,7 @@ dependencies = [
  "hex",
  "hmac",
  "hyper 0.14.30",
- "hyper 1.4.1",
+ "hyper 1.6.0",
  "hyper-util",
  "indexmap 2.10.0",
  "itertools 0.12.1",
@@ -9087,7 +9530,7 @@ dependencies = [
  "subtle",
  "syn 2.0.100",
  "sync_wrapper 0.1.2",
- "thiserror 2.0.11",
+ "thiserror 2.0.12",
  "tikv-jemalloc-ctl",
  "tikv-jemalloc-sys",
  "time",
@@ -9098,7 +9541,7 @@ dependencies = [
  "tokio-util",
  "toml_edit",
  "tonic",
- "tower 0.5.2",
+ "tower",
  "tracing",
  "tracing-core",
  "tracing-log",
@@ -9155,11 +9598,12 @@ dependencies = [
 
 [[package]]
 name = "xattr"
-version = "1.0.0"
+version = "1.5.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ea263437ca03c1522846a4ddafbca2542d0ad5ed9b784909d4b27b76f62bc34a"
+checksum = "af3a19837351dc82ba89f8a125e22a3c475f05aba604acc023d62b2739ae2909"
 dependencies = [
  "libc",
+ "rustix 1.0.8",
 ]
 
 [[package]]
diff --git a/storage_controller/Cargo.toml b/storage_controller/Cargo.toml
index d67be6d469..e1088b91e4 100644
--- a/storage_controller/Cargo.toml
+++ b/storage_controller/Cargo.toml
@@ -74,4 +74,8 @@ http-utils = { path = "../libs/http-utils/" }
 utils = { path = "../libs/utils/" }
 metrics = { path = "../libs/metrics/" }
 control_plane = { path = "../control_plane" }
-workspace_hack = { version = "0.1", path = "../workspace_hack" }
\ No newline at end of file
+workspace_hack = { version = "0.1", path = "../workspace_hack" }
+
+[dev-dependencies]
+postgresql_archive = "0.19.0"
+postgresql_embedded = { version = "0.19.0", features = ["blocking"] }
diff --git a/storage_controller/src/hadron_dns.rs b/storage_controller/src/hadron_dns.rs
new file mode 100644
index 0000000000..f148fbe9e5
--- /dev/null
+++ b/storage_controller/src/hadron_dns.rs
@@ -0,0 +1,7 @@
+/// Type of the storage node (pageserver or safekeeper) that we are updating DNS records for. Different types of nodes will have
+/// different-looking DNS names in the DNS zone.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+pub enum NodeType {
+    Pageserver,
+    Safekeeper,
+}
diff --git a/storage_controller/src/hadron_queries.rs b/storage_controller/src/hadron_queries.rs
new file mode 100644
index 0000000000..71dd5dcee2
--- /dev/null
+++ b/storage_controller/src/hadron_queries.rs
@@ -0,0 +1,848 @@
+use std::collections::{HashMap, HashSet};
+
+use diesel::Queryable;
+use diesel::dsl::min;
+use diesel::prelude::*;
+use diesel_async::AsyncConnection;
+use diesel_async::AsyncPgConnection;
+use diesel_async::RunQueryDsl;
+use itertools::Itertools;
+use pageserver_api::controller_api::SCSafekeeperTimelinesResponse;
+use scoped_futures::ScopedFutureExt;
+use serde::{Deserialize, Serialize};
+use utils::id::{NodeId, TenantId, TimelineId};
+use uuid::Uuid;
+
+use crate::hadron_dns::NodeType;
+use crate::hadron_requests::NodeConnectionInfo;
+use crate::persistence::{DatabaseError, DatabaseResult};
+use crate::schema::{hadron_safekeepers, nodes};
+use crate::sk_node::SafeKeeperNode;
+use std::str::FromStr;
+
+// The Safe Keeper node database representation (for Diesel).
+#[derive(
+    Clone, Serialize, Deserialize, Queryable, Selectable, Insertable, Eq, PartialEq, AsChangeset,
+)]
+#[diesel(table_name = crate::schema::hadron_safekeepers)]
+pub(crate) struct HadronSafekeeperRow {
+    pub(crate) sk_node_id: i64,
+    pub(crate) listen_http_addr: String,
+    pub(crate) listen_http_port: i32,
+    pub(crate) listen_pg_addr: String,
+    pub(crate) listen_pg_port: i32,
+}
+
+#[derive(
+    Clone, Serialize, Deserialize, Queryable, Selectable, Insertable, Eq, PartialEq, AsChangeset,
+)]
+#[diesel(table_name = crate::schema::hadron_timeline_safekeepers)]
+pub(crate) struct HadronTimelineSafekeeper {
+    pub(crate) timeline_id: String,
+    pub(crate) sk_node_id: i64,
+    pub(crate) legacy_endpoint_id: Option<Uuid>,
+}
+
+pub async fn execute_sk_upsert(
+    conn: &mut AsyncPgConnection,
+    sk_row: HadronSafekeeperRow,
+) -> DatabaseResult<()> {
+    // SQL:
+    // INSERT INTO hadron_safekeepers (sk_node_id, listen_http_addr, listen_http_port, listen_pg_addr, listen_pg_port)
+    // VALUES ($1, $2, $3, $4, $5)
+    // ON CONFLICT (sk_node_id)
+    // DO UPDATE SET listen_http_addr = $2, listen_http_port = $3, listen_pg_addr = $4, listen_pg_port = $5;
+
+    use crate::schema::hadron_safekeepers::dsl::*;
+
+    diesel::insert_into(hadron_safekeepers)
+        .values(&sk_row)
+        .on_conflict(sk_node_id)
+        .do_update()
+        .set(&sk_row)
+        .execute(conn)
+        .await?;
+
+    Ok(())
+}
+
+// Load all safekeeper nodes and their associated timelines from the meta PG. This query is supposed
+// to run only once on HCC startup and is used to construct the SafeKeeperScheduler state. Performs
+// scans of the hadron_safekeepers and hadron_timeline_safekeepers tables.
+pub async fn scan_safekeepers_and_scheduled_timelines(
+    conn: &mut AsyncPgConnection,
+) -> DatabaseResult<HashMap<NodeId, SafeKeeperNode>> {
+    use crate::schema::hadron_safekeepers;
+    use crate::schema::hadron_timeline_safekeepers;
+
+    // We first scan the hadron_safekeepers table to constuct the SafeKeeperNode objects. We don't know anything about
+    // the timelines scheduled to the safekeepers after this step. We then scan the hadron_timeline_safekeepers table
+    // to populate the data structures in the SafeKeeperNode objects to reflect the timelines scheduled to the safekeepers.
+    let mut results: HashMap<NodeId, SafeKeeperNode> = hadron_safekeepers::table
+        .select((
+            hadron_safekeepers::sk_node_id,
+            hadron_safekeepers::listen_http_addr,
+            hadron_safekeepers::listen_http_port,
+            hadron_safekeepers::listen_pg_addr,
+            hadron_safekeepers::listen_pg_port,
+        ))
+        .load::<HadronSafekeeperRow>(conn)
+        .await?
+        .into_iter()
+        .map(|row| {
+            let sk_node = SafeKeeperNode {
+                id: NodeId(row.sk_node_id as u64),
+                listen_http_addr: row.listen_http_addr.clone(),
+                listen_http_port: row.listen_http_port as u16,
+                listen_pg_addr: row.listen_pg_addr.clone(),
+                listen_pg_port: row.listen_pg_port as u16,
+                legacy_endpoints: HashMap::new(),
+                timelines: HashSet::new(),
+            };
+            (sk_node.id, sk_node)
+        })
+        .collect();
+
+    let timeline_sk_rows = hadron_timeline_safekeepers::table
+        .select((
+            hadron_timeline_safekeepers::sk_node_id,
+            hadron_timeline_safekeepers::timeline_id,
+            hadron_timeline_safekeepers::legacy_endpoint_id,
+        ))
+        .load::<(i64, String, Option<Uuid>)>(conn)
+        .await?;
+    for (sk_node_id, timeline_id, legacy_endpoint_id) in timeline_sk_rows {
+        if let Some(sk_node) = results.get_mut(&NodeId(sk_node_id as u64)) {
+            let parsed_timeline_id =
+                TimelineId::from_str(&timeline_id).map_err(|e: hex::FromHexError| {
+                    DatabaseError::Logical(format!("Failed to parse timeline IDs: {e}"))
+                })?;
+            sk_node.timelines.insert(parsed_timeline_id);
+            if let Some(legacy_endpoint_id) = legacy_endpoint_id {
+                sk_node
+                    .legacy_endpoints
+                    .insert(legacy_endpoint_id, parsed_timeline_id);
+            }
+        }
+    }
+
+    Ok(results)
+}
+
+// Queries the hadron_timeline_safekeepers table to get the safekeepers assigned to the passed
+// timeline. If none are found, persists the input proposed safekeepers to the table and returns
+// them.
+pub async fn idempotently_persist_or_get_existing_timeline_safekeepers(
+    conn: &mut AsyncPgConnection,
+    timeline_id: TimelineId,
+    safekeepers: &[NodeId],
+) -> DatabaseResult<Vec<NodeId>> {
+    use crate::schema::hadron_timeline_safekeepers;
+    // Confirm and persist the timeline-safekeeper mapping. If there are existing safekeepers
+    // assigned to the timeline in the database, treat those as the source of truth.
+    let existing_safekeepers: Vec<i64> = hadron_timeline_safekeepers::table
+        .select(hadron_timeline_safekeepers::sk_node_id)
+        .filter(hadron_timeline_safekeepers::timeline_id.eq(timeline_id.to_string()))
+        .load::<i64>(conn)
+        .await?;
+    let confirmed_safekeepers: Vec<NodeId> = if existing_safekeepers.is_empty() {
+        let proposed_safekeeper_endpoint_rows_result: Result<Vec<HadronTimelineSafekeeper>, _> =
+            safekeepers
+                .iter()
+                .map(|sk_node_id| {
+                    i64::try_from(sk_node_id.0).map(|sk_node_id| HadronTimelineSafekeeper {
+                        timeline_id: timeline_id.to_string(),
+                        sk_node_id,
+                        legacy_endpoint_id: None,
+                    })
+                })
+                .collect();
+
+        let proposed_safekeeper_endpoint_rows =
+            proposed_safekeeper_endpoint_rows_result.map_err(|e| {
+                DatabaseError::Logical(format!("Failed to convert safekeeper IDs: {e}"))
+            })?;
+
+        diesel::insert_into(hadron_timeline_safekeepers::table)
+            .values(&proposed_safekeeper_endpoint_rows)
+            .execute(conn)
+            .await?;
+        safekeepers.to_owned()
+    } else {
+        let safekeeper_result: Result<Vec<NodeId>, _> = existing_safekeepers
+            .into_iter()
+            .map(|arg0: i64| u64::try_from(arg0).map(NodeId))
+            .collect();
+
+        safekeeper_result
+            .map_err(|e| DatabaseError::Logical(format!("Failed to convert safekeeper IDs: {e}")))?
+    };
+
+    Ok(confirmed_safekeepers)
+}
+
+pub async fn delete_timeline_safekeepers(
+    conn: &mut AsyncPgConnection,
+    timeline_id: TimelineId,
+) -> DatabaseResult<()> {
+    use crate::schema::hadron_timeline_safekeepers;
+
+    diesel::delete(hadron_timeline_safekeepers::table)
+        .filter(hadron_timeline_safekeepers::timeline_id.eq(timeline_id.to_string()))
+        .execute(conn)
+        .await?;
+
+    Ok(())
+}
+
+pub(crate) async fn execute_safekeeper_list_timelines(
+    conn: &mut AsyncPgConnection,
+    safekeeper_id: i64,
+) -> DatabaseResult<SCSafekeeperTimelinesResponse> {
+    use crate::schema::hadron_timeline_safekeepers;
+    use pageserver_api::controller_api::SCSafekeeperTimelinesResponse;
+
+    conn.transaction(|conn| {
+        async move {
+            let mut sk_timelines = SCSafekeeperTimelinesResponse {
+                timelines: Vec::new(),
+                safekeeper_peers: Vec::new(),
+            };
+
+            // Find all timelines <String>
+            let timeline_ids = hadron_timeline_safekeepers::table
+                .select(hadron_timeline_safekeepers::timeline_id)
+                .filter(hadron_timeline_safekeepers::sk_node_id.eq(safekeeper_id))
+                .load::<String>(conn)
+                .await
+                .into_iter()
+                .flatten()
+                .collect_vec();
+
+            // Find the peers for each timeline. <timeline_id, sk_node_id>
+            let timeline_peers = hadron_timeline_safekeepers::table
+                .select((
+                    hadron_timeline_safekeepers::timeline_id,
+                    hadron_timeline_safekeepers::sk_node_id,
+                ))
+                .filter(hadron_timeline_safekeepers::timeline_id.eq_any(&timeline_ids))
+                .load::<(String, i64)>(conn)
+                .await
+                .into_iter()
+                .flatten()
+                .collect_vec();
+
+            let mut timeline_peers_map = HashMap::new();
+            let mut seen = HashSet::new();
+            let mut unique_sks = Vec::new();
+
+            for (timeline_id, sk_node_id) in timeline_peers {
+                timeline_peers_map
+                    .entry(timeline_id)
+                    .or_insert_with(Vec::new)
+                    .push(sk_node_id);
+                if seen.insert(sk_node_id) {
+                    unique_sks.push(sk_node_id);
+                }
+            }
+
+            // Find SK info.
+            let mut found_sk_nodes = HashSet::new();
+            hadron_safekeepers::table
+                .select((
+                    hadron_safekeepers::sk_node_id,
+                    hadron_safekeepers::listen_http_addr,
+                    hadron_safekeepers::listen_http_port,
+                ))
+                .filter(hadron_safekeepers::sk_node_id.eq_any(&unique_sks))
+                .load::<(i64, String, i32)>(conn)
+                .await
+                .into_iter()
+                .flatten()
+                .for_each(|(sk_node_id, listen_http_addr, http_port)| {
+                    found_sk_nodes.insert(sk_node_id);
+
+                    sk_timelines.safekeeper_peers.push(
+                        pageserver_api::controller_api::TimelineSafekeeperPeer {
+                            node_id: utils::id::NodeId(sk_node_id as u64),
+                            listen_http_addr,
+                            http_port,
+                        },
+                    );
+                });
+
+            // Prepare timeline response.
+            for timeline_id in timeline_ids {
+                if !timeline_peers_map.contains_key(&timeline_id) {
+                    continue;
+                }
+                let peers = timeline_peers_map.get(&timeline_id).unwrap();
+                // Check peers exist.
+                if !peers
+                    .iter()
+                    .all(|sk_node_id| found_sk_nodes.contains(sk_node_id))
+                {
+                    continue;
+                }
+
+                let timeline = pageserver_api::controller_api::SCSafekeeperTimeline {
+                    timeline_id: TimelineId::from_str(&timeline_id).unwrap(),
+                    peers: peers
+                        .iter()
+                        .map(|sk_node_id| utils::id::NodeId(*sk_node_id as u64))
+                        .collect(),
+                };
+                sk_timelines.timelines.push(timeline);
+            }
+
+            Ok(sk_timelines)
+        }
+        .scope_boxed()
+    })
+    .await
+}
+
+/// Stores details about connecting to pageserver and safekeeper nodes for a given tenant and
+/// timeline.
+pub struct PageserverAndSafekeeperConnectionInfo {
+    pub pageserver_conn_info: Vec<NodeConnectionInfo>,
+    pub safekeeper_conn_info: Vec<NodeConnectionInfo>,
+}
+
+/// Retrieves the connection information for the pageserver and safekeepers associated with the
+/// given tenant and timeline.
+pub async fn get_pageserver_and_safekeeper_connection_info(
+    conn: &mut AsyncPgConnection,
+    tenant_id: TenantId,
+    timeline_id: TimelineId,
+) -> DatabaseResult<PageserverAndSafekeeperConnectionInfo> {
+    conn.transaction(|conn| {
+        async move {
+            // Fetch details about pageserver, which is associated with the input tenant.
+            let pageserver_conn_info =
+                get_pageserver_connection_info(conn, &tenant_id.to_string()).await?;
+
+            // Fetch details about safekeepers, which are associated with the input timeline.
+            let safekeeper_conn_info =
+                get_safekeeper_connection_info(conn, &timeline_id.to_string()).await?;
+
+            Ok(PageserverAndSafekeeperConnectionInfo {
+                pageserver_conn_info,
+                safekeeper_conn_info,
+            })
+        }
+        .scope_boxed()
+    })
+    .await
+}
+
+async fn get_safekeeper_connection_info(
+    conn: &mut AsyncPgConnection,
+    timeline_id: &str,
+) -> DatabaseResult<Vec<NodeConnectionInfo>> {
+    use crate::schema::hadron_safekeepers;
+    use crate::schema::hadron_timeline_safekeepers;
+
+    Ok(hadron_timeline_safekeepers::table
+        .inner_join(
+            hadron_safekeepers::table
+                .on(hadron_timeline_safekeepers::sk_node_id.eq(hadron_safekeepers::sk_node_id)),
+        )
+        .select((
+            hadron_safekeepers::sk_node_id,
+            hadron_safekeepers::listen_pg_addr,
+            hadron_safekeepers::listen_pg_port,
+        ))
+        .filter(hadron_timeline_safekeepers::timeline_id.eq(timeline_id.to_string()))
+        .load::<(i64, String, i32)>(conn)
+        .await?
+        .into_iter()
+        .map(|(node_id, addr, port)| {
+            NodeConnectionInfo::new(
+                NodeType::Safekeeper,
+                NodeId(node_id as u64),
+                addr,
+                port as u16,
+            )
+        })
+        .collect())
+}
+
+async fn get_pageserver_connection_info(
+    conn: &mut AsyncPgConnection,
+    tenant_id: &str,
+) -> DatabaseResult<Vec<NodeConnectionInfo>> {
+    use crate::schema::tenant_shards;
+
+    // When the tenant is being split, it'll contain both old shards and new shards. Until the tenant split is committed,
+    // we should always use the old shards.
+    // NOTE: we only support tenant split without tennat merge. Thus shard count could only increase.
+    let min_shard_count = match tenant_shards::table
+        .select(min(tenant_shards::shard_count))
+        .filter(tenant_shards::tenant_id.eq(tenant_id))
+        .first::<Option<i32>>(conn)
+        .await
+        .optional()?
+    {
+        Some(Some(count)) => count,
+        Some(None) => {
+            // Tenant doesn't exist. It's possible that it was deleted before we got the request.
+            return Ok(vec![]);
+        }
+        None => {
+            // This is never supposed to happen because `SELECT min()` should always return one row.
+            return Err(DatabaseError::Logical(format!(
+                "Unexpected empty query result for min(shard_count) query. Tenant ID {}",
+                tenant_id
+            )));
+        }
+    };
+
+    let shards: Vec<NodeConnectionInfo> = nodes::table
+        .inner_join(
+            tenant_shards::table.on(nodes::node_id
+                .nullable()
+                .eq(tenant_shards::generation_pageserver)),
+        )
+        .select((nodes::node_id, nodes::listen_pg_addr, nodes::listen_pg_port))
+        .filter(tenant_shards::tenant_id.eq(&tenant_id.to_string()))
+        .order(tenant_shards::shard_number.asc())
+        .filter(tenant_shards::shard_count.eq(min_shard_count))
+        .load::<(i64, String, i32)>(conn)
+        .await?
+        .into_iter()
+        .map(|(node_id, addr, port)| {
+            NodeConnectionInfo::new(
+                NodeType::Pageserver,
+                NodeId(node_id as u64),
+                addr,
+                port as u16,
+            )
+        })
+        .collect();
+
+    if !shards.is_empty() && !shards.len().is_power_of_two() {
+        return Err(DatabaseError::Logical(format!(
+            "Tenant {} has unexpected shard count {} (not a power of 2)",
+            tenant_id,
+            shards.len()
+        )));
+    }
+    Ok(shards)
+}
+
+#[cfg(test)]
+mod test {
+
+    use std::collections::BTreeMap;
+
+    use super::*;
+    use crate::schema::hadron_safekeepers;
+    use diesel::PgConnection;
+    use diesel_migrations::{EmbeddedMigrations, MigrationHarness, embed_migrations};
+    use pageserver_api::controller_api::{SCSafekeeperTimeline, TimelineSafekeeperPeer};
+    use postgresql_archive::VersionReq;
+    use postgresql_embedded::Settings;
+    use postgresql_embedded::blocking::PostgreSQL;
+
+    async fn get_embedded_pg() -> postgresql_embedded::Result<PostgreSQL> {
+        tokio::task::spawn_blocking(|| {
+            let pg_install_dir = "../pg_install/16.0.0";
+            // Link "pg_install/v16" -> "pg_install/16.0.0" so that it can be picked up by the postgres_embedded
+            // crate without needing to download anything. The postgres_embedded crate expects a specific format
+            // for the directory name.
+            let _ = std::os::unix::fs::symlink("./v16", pg_install_dir);
+
+            let settings = Settings {
+                installation_dir: std::path::PathBuf::from(pg_install_dir),
+                username: "postgres".to_string(),
+                password: "password".to_string(),
+                // Use a 30-second timeout for database initialization to avoid flakiness in the CI environment.
+                timeout: Some(std::time::Duration::from_secs(30)),
+                version: VersionReq::parse("=16.0.0").unwrap(),
+                ..Default::default()
+            };
+            let mut pg = PostgreSQL::new(settings);
+
+            pg.setup()?;
+            pg.start()?;
+
+            pg.create_database("test")?;
+
+            Ok(pg)
+        })
+        .await
+        .unwrap()
+    }
+
+    pub const MIGRATIONS: EmbeddedMigrations = embed_migrations!("migrations");
+
+    fn run_migrations(connection: &mut PgConnection) -> Result<(), String> {
+        connection.run_pending_migrations(MIGRATIONS).unwrap();
+        Ok(())
+    }
+
+    fn get_test_sk_node(id: u64) -> SafeKeeperNode {
+        SafeKeeperNode::new(
+            NodeId(id),
+            format!("safekeeper-{}", id),
+            123,
+            format!("safekeeper-{}", id),
+            456,
+        )
+    }
+
+    #[tokio::test]
+    async fn test_safekeeper_upserts_and_list() {
+        let pg = get_embedded_pg().await.unwrap();
+
+        let connection_string = pg.settings().url("test");
+        {
+            let mut conn = PgConnection::establish(&connection_string)
+                .unwrap_or_else(|_| panic!("Error connecting to {}", connection_string));
+            run_migrations(&mut conn).unwrap();
+        }
+
+        let mut connection = AsyncPgConnection::establish(&connection_string)
+            .await
+            .unwrap_or_else(|_| panic!("Error connecting to {}", connection_string));
+
+        execute_sk_upsert(&mut connection, get_test_sk_node(0).to_database_row())
+            .await
+            .unwrap();
+        execute_sk_upsert(&mut connection, get_test_sk_node(1).to_database_row())
+            .await
+            .unwrap();
+        execute_sk_upsert(&mut connection, get_test_sk_node(2).to_database_row())
+            .await
+            .unwrap();
+
+        // Insert an entry into the hadron_timeline_safekeepers table.
+        use crate::schema::hadron_timeline_safekeepers;
+        let timeline1_id = TimelineId::generate();
+        diesel::insert_into(hadron_timeline_safekeepers::table)
+            .values(&HadronTimelineSafekeeper {
+                timeline_id: timeline1_id.to_string(),
+                sk_node_id: 0,
+                legacy_endpoint_id: None,
+            })
+            .execute(&mut connection)
+            .await
+            .expect("Failed to insert timeline1");
+
+        // Test that the nodes have indeed been inserted
+        let sk_nodes = hadron_safekeepers::table
+            .load::<HadronSafekeeperRow>(&mut connection)
+            .await
+            .unwrap();
+        assert_eq!(sk_nodes.len(), 3);
+        assert_eq!(sk_nodes[0].sk_node_id, 0);
+        assert_eq!(sk_nodes[1].sk_node_id, 1);
+        assert_eq!(sk_nodes[2].sk_node_id, 2);
+
+        // Test that we can read the nodes back out in the join query, where we pull all the Safekeepers along with their endpoints scheduled.
+        // There should be no endpoints in this test, verify that nothing breaks.
+        let sk_nodes = scan_safekeepers_and_scheduled_timelines(&mut connection)
+            .await
+            .unwrap();
+        assert_eq!(sk_nodes.len(), 3);
+        assert_eq!(sk_nodes[&NodeId(0)].legacy_endpoints.len(), 0);
+        assert_eq!(sk_nodes[&NodeId(1)].legacy_endpoints.len(), 0);
+        assert_eq!(sk_nodes[&NodeId(2)].legacy_endpoints.len(), 0);
+
+        // Test that only the 0th safekeeper is assigned to the timeline.
+        assert_eq!(sk_nodes[&NodeId(0)].timelines.len(), 1);
+        assert_eq!(sk_nodes[&NodeId(1)].timelines.len(), 0);
+        assert_eq!(sk_nodes[&NodeId(2)].timelines.len(), 0);
+    }
+
+    #[tokio::test]
+    async fn test_idempotently_persist_or_get_existing_timeline_safekeepers() {
+        let pg = get_embedded_pg().await.unwrap();
+
+        let connection_string = pg.settings().url("test");
+        {
+            let mut conn = PgConnection::establish(&connection_string)
+                .unwrap_or_else(|_| panic!("Error connecting to {}", connection_string));
+            run_migrations(&mut conn).unwrap();
+        }
+
+        let mut connection = AsyncPgConnection::establish(&connection_string)
+            .await
+            .unwrap_or_else(|_| panic!("Error connecting to {}", connection_string));
+
+        // An initial call should insert the timeline safekeepers and return the inserted values.
+        let timeline1_id = TimelineId::generate();
+        let safekeeper_ids = vec![NodeId(1), NodeId(2), NodeId(3)];
+        let inserted = idempotently_persist_or_get_existing_timeline_safekeepers(
+            &mut connection,
+            timeline1_id,
+            &safekeeper_ids,
+        )
+        .await
+        .expect("Failed to insert timeline safekeepers");
+
+        assert_eq!(inserted, safekeeper_ids);
+
+        // A second call with the same timeline should return the same safekeeper IDs.
+        let retrieved = idempotently_persist_or_get_existing_timeline_safekeepers(
+            &mut connection,
+            timeline1_id,
+            &[NodeId(4), NodeId(5), NodeId(6)],
+        )
+        .await
+        .expect("Failed to retrieve timeline safekeepers");
+
+        assert_eq!(retrieved, safekeeper_ids);
+    }
+
+    async fn load_timelines_by_sk_node(
+        conn: &mut AsyncPgConnection,
+    ) -> DatabaseResult<BTreeMap<i64, Vec<String>>> {
+        use crate::schema::hadron_timeline_safekeepers;
+
+        let rows = hadron_timeline_safekeepers::table
+            .select((
+                hadron_timeline_safekeepers::sk_node_id,
+                hadron_timeline_safekeepers::timeline_id,
+            ))
+            .load::<(i64, String)>(conn)
+            .await?;
+
+        let mut timelines_by_sk_node = BTreeMap::new();
+        for (sk_node_id, timeline_id) in rows {
+            timelines_by_sk_node
+                .entry(sk_node_id)
+                .or_insert_with(Vec::new)
+                .push(timeline_id);
+        }
+
+        Ok(timelines_by_sk_node)
+    }
+
+    #[tokio::test]
+    async fn test_delete_timeline_safekeepers() {
+        let pg = get_embedded_pg().await.unwrap();
+
+        let connection_string = pg.settings().url("test");
+        {
+            let mut conn = PgConnection::establish(&connection_string)
+                .unwrap_or_else(|_| panic!("Error connecting to {}", connection_string));
+            run_migrations(&mut conn).unwrap();
+        }
+
+        let mut connection = AsyncPgConnection::establish(&connection_string)
+            .await
+            .unwrap_or_else(|_| panic!("Error connecting to {}", connection_string));
+
+        // Insert some values
+        let timeline1_id = TimelineId::generate();
+        let safekeeper_ids = vec![NodeId(1), NodeId(2), NodeId(3)];
+        idempotently_persist_or_get_existing_timeline_safekeepers(
+            &mut connection,
+            timeline1_id,
+            &safekeeper_ids,
+        )
+        .await
+        .expect("Failed to insert timeline safekeepers");
+
+        // Validate that the values were inserted
+        let inserted = load_timelines_by_sk_node(&mut connection)
+            .await
+            .expect("Failed to load timelines by sk node");
+
+        assert_eq!(inserted.get(&1).unwrap().len(), 1);
+        assert_eq!(inserted.get(&2).unwrap().len(), 1);
+        assert_eq!(inserted.get(&3).unwrap().len(), 1);
+
+        // Delete the values
+        delete_timeline_safekeepers(&mut connection, timeline1_id)
+            .await
+            .expect("Failed to delete timeline safekeepers");
+
+        // Validate that the values were deleted
+        let deleted = load_timelines_by_sk_node(&mut connection)
+            .await
+            .expect("Failed to load timelines by sk node");
+
+        assert!(deleted.is_empty());
+    }
+
+    fn assert_list_safekeeper_timelines(
+        actual: &mut SCSafekeeperTimelinesResponse,
+        expected: &mut SCSafekeeperTimelinesResponse,
+    ) {
+        assert_eq!(actual.timelines.len(), expected.timelines.len());
+        assert_eq!(
+            actual.safekeeper_peers.len(),
+            expected.safekeeper_peers.len()
+        );
+
+        actual.timelines.sort_by_key(|item| item.timeline_id);
+        expected.timelines.sort_by_key(|item| item.timeline_id);
+
+        actual.safekeeper_peers.sort_by_key(|item| item.node_id);
+        expected.safekeeper_peers.sort_by_key(|item| item.node_id);
+
+        for i in 0..actual.timelines.len() {
+            let mut at = actual.timelines[i].clone();
+            let mut et = expected.timelines[i].clone();
+            at.peers.sort_by_key(|item| item.0);
+            et.peers.sort_by_key(|item| item.0);
+
+            assert_eq!(at.timeline_id, et.timeline_id);
+
+            assert!(
+                at.peers.iter().eq(et.peers.iter()),
+                "at peers: {:#?}, et peers: {:#?}",
+                at.peers,
+                et.peers
+            );
+        }
+
+        for i in 0..actual.safekeeper_peers.len() {
+            let at = actual.safekeeper_peers[i].clone();
+            let et = expected.safekeeper_peers[i].clone();
+            assert_eq!(at.node_id, et.node_id);
+            assert_eq!(at.listen_http_addr, et.listen_http_addr);
+            assert_eq!(at.http_port, et.http_port);
+        }
+    }
+
+    #[tokio::test]
+    async fn test_list_safekeeper_timelines() {
+        let pg = get_embedded_pg().await.unwrap();
+
+        let connection_string = pg.settings().url("test");
+        {
+            let mut conn = PgConnection::establish(&connection_string)
+                .unwrap_or_else(|_| panic!("Error connecting to {}", connection_string));
+            run_migrations(&mut conn).unwrap();
+        }
+        let mut connection = AsyncPgConnection::establish(&connection_string)
+            .await
+            .unwrap_or_else(|_| panic!("Error connecting to {}", connection_string));
+
+        // Insert some values
+        let safekeeper_ids = vec![
+            NodeId(0),
+            NodeId(1),
+            NodeId(2),
+            NodeId(3),
+            NodeId(4),
+            NodeId(5),
+        ];
+        for safekeeper_id in &safekeeper_ids {
+            execute_sk_upsert(
+                &mut connection,
+                get_test_sk_node(safekeeper_id.0).to_database_row(),
+            )
+            .await
+            .unwrap();
+        }
+
+        // Create some endpoints.
+        // 5 use SK-0/1/2
+        // 5 use SK-2/3/4
+        let mut timeline_ids = Vec::new();
+
+        for i in 0..10 {
+            let timeline_id = TimelineId::generate();
+            timeline_ids.push(timeline_id);
+
+            let safekeepers = if i < 5 {
+                vec![NodeId(0), NodeId(1), NodeId(2)]
+            } else {
+                vec![NodeId(2), NodeId(3), NodeId(4)]
+            };
+
+            idempotently_persist_or_get_existing_timeline_safekeepers(
+                &mut connection,
+                timeline_id,
+                &safekeepers,
+            )
+            .await
+            .unwrap();
+        }
+
+        // SK-0/1 owns the first 5 timelines.
+        // SK-2 owns all 10 timelines.
+        // SK-3/4 owns the last 5 timelines.
+        // SK-5 owns no timelines.
+        // SK-6 does not exist.
+        let mut expected_responses = vec![
+            SCSafekeeperTimelinesResponse {
+                timelines: Vec::new(),
+                safekeeper_peers: Vec::new(),
+            };
+            7
+        ];
+
+        // SC does not know the tenant ids.
+        for (i, timeline_id) in timeline_ids.iter().enumerate().take(10) {
+            if i < 5 {
+                expected_responses[0].timelines.push(SCSafekeeperTimeline {
+                    timeline_id: *timeline_id,
+                    peers: vec![NodeId(0), NodeId(1), NodeId(2)],
+                });
+
+                expected_responses[2].timelines.push(SCSafekeeperTimeline {
+                    timeline_id: *timeline_id,
+                    peers: vec![NodeId(0), NodeId(1), NodeId(2)],
+                });
+                continue;
+            }
+
+            expected_responses[2].timelines.push(SCSafekeeperTimeline {
+                timeline_id: *timeline_id,
+                peers: vec![NodeId(2), NodeId(3), NodeId(4)],
+            });
+            expected_responses[3].timelines.push(SCSafekeeperTimeline {
+                timeline_id: *timeline_id,
+                peers: vec![NodeId(2), NodeId(3), NodeId(4)],
+            });
+        }
+        for i in 0..5 {
+            expected_responses[2]
+                .safekeeper_peers
+                .push(TimelineSafekeeperPeer {
+                    node_id: NodeId(i),
+                    listen_http_addr: format!("safekeeper-{}", i),
+                    http_port: 123,
+                });
+            if i < 3 {
+                expected_responses[0]
+                    .safekeeper_peers
+                    .push(TimelineSafekeeperPeer {
+                        node_id: NodeId(i),
+                        listen_http_addr: format!("safekeeper-{}", i),
+                        http_port: 123,
+                    });
+                expected_responses[3]
+                    .safekeeper_peers
+                    .push(TimelineSafekeeperPeer {
+                        node_id: NodeId(i + 2),
+                        listen_http_addr: format!("safekeeper-{}", i + 2),
+                        http_port: 123,
+                    });
+            }
+        }
+        expected_responses[1] = expected_responses[0].clone();
+        expected_responses[4] = expected_responses[3].clone();
+
+        for safekeeper_id in &safekeeper_ids {
+            let sk_timelines: Result<SCSafekeeperTimelinesResponse, DatabaseError> =
+                execute_safekeeper_list_timelines(
+                    &mut connection,
+                    safekeeper_id.0.try_into().unwrap(),
+                )
+                .await;
+            assert!(sk_timelines.is_ok());
+            let mut sk_timelines: SCSafekeeperTimelinesResponse = sk_timelines.unwrap();
+            assert_list_safekeeper_timelines(
+                &mut sk_timelines,
+                &mut expected_responses[safekeeper_id.0 as usize],
+            );
+        }
+    }
+}
diff --git a/storage_controller/src/hadron_requests.rs b/storage_controller/src/hadron_requests.rs
new file mode 100644
index 0000000000..980cfee306
--- /dev/null
+++ b/storage_controller/src/hadron_requests.rs
@@ -0,0 +1,34 @@
+use utils::id::NodeId;
+
+use crate::hadron_dns::NodeType;
+
+/// Internal representation of how a compute node should connect to a PS or SK node. HCC uses this struct to
+/// construct connection strings that are passed to the compute node via the compute spec. This struct is never
+/// serialized or sent over the wire.
+#[derive(Debug, PartialEq, Eq, PartialOrd, Ord)]
+pub(crate) struct NodeConnectionInfo {
+    // Type of the node.
+    node_type: NodeType,
+    // Node ID. Unique for each node type.
+    pub(crate) node_id: NodeId,
+    // The hostname reported by the node when it registers. This is the hostname we store in the meta PG, and is
+    // typically the k8s cluster DNS name of the node. Note that this may not be resolvable from compute nodes running
+    // on dblet. For this reason, this hostname is usually not communicated to the compute node. Instead, HCC computes
+    // a DNS name of the node in the Cloud DNS hosted zone based on `node_type` and `node_id` and advertise the DNS name
+    // to compute nodes. This hostname here is used as a fallback in tests or other scenarios where we do not have the
+    // Cloud DNS hosted zone available.
+    registration_hostname: String,
+    // The PG wire protocol port on the PS or SK node.
+    port: u16,
+}
+
+impl NodeConnectionInfo {
+    pub(crate) fn new(node_type: NodeType, node_id: NodeId, hostname: String, port: u16) -> Self {
+        NodeConnectionInfo {
+            node_type,
+            node_id,
+            registration_hostname: hostname,
+            port,
+        }
+    }
+}
diff --git a/storage_controller/src/lib.rs b/storage_controller/src/lib.rs
index 24b06da83a..762f2dc46b 100644
--- a/storage_controller/src/lib.rs
+++ b/storage_controller/src/lib.rs
@@ -6,6 +6,9 @@ extern crate hyper0 as hyper;
 mod auth;
 mod background_node_operations;
 mod compute_hook;
+pub mod hadron_dns;
+mod hadron_queries;
+pub mod hadron_requests;
 pub mod hadron_utils;
 mod heartbeater;
 pub mod http;
@@ -23,6 +26,7 @@ mod safekeeper_client;
 mod scheduler;
 mod schema;
 pub mod service;
+mod sk_node;
 mod tenant_shard;
 mod timeline_import;
 
diff --git a/storage_controller/src/persistence.rs b/storage_controller/src/persistence.rs
index 619b5f69b8..17f9d19697 100644
--- a/storage_controller/src/persistence.rs
+++ b/storage_controller/src/persistence.rs
@@ -20,7 +20,8 @@ use futures::future::BoxFuture;
 use itertools::Itertools;
 use pageserver_api::controller_api::{
     AvailabilityZone, MetadataHealthRecord, NodeLifecycle, NodeSchedulingPolicy, PlacementPolicy,
-    SafekeeperDescribeResponse, ShardSchedulingPolicy, SkSchedulingPolicy,
+    SCSafekeeperTimelinesResponse, SafekeeperDescribeResponse, ShardSchedulingPolicy,
+    SkSchedulingPolicy,
 };
 use pageserver_api::models::{ShardImportStatus, TenantConfig};
 use pageserver_api::shard::{
@@ -37,10 +38,19 @@ use utils::id::{NodeId, TenantId, TimelineId};
 use utils::lsn::Lsn;
 
 use self::split_state::SplitState;
+use crate::hadron_queries::HadronSafekeeperRow;
+use crate::hadron_queries::PageserverAndSafekeeperConnectionInfo;
+use crate::hadron_queries::delete_timeline_safekeepers;
+use crate::hadron_queries::execute_safekeeper_list_timelines;
+use crate::hadron_queries::execute_sk_upsert;
+use crate::hadron_queries::get_pageserver_and_safekeeper_connection_info;
+use crate::hadron_queries::idempotently_persist_or_get_existing_timeline_safekeepers;
+use crate::hadron_queries::scan_safekeepers_and_scheduled_timelines;
 use crate::metrics::{
     DatabaseQueryErrorLabelGroup, DatabaseQueryLatencyLabelGroup, METRICS_REGISTRY,
 };
 use crate::node::Node;
+use crate::sk_node::SafeKeeperNode;
 use crate::timeline_import::{
     TimelineImport, TimelineImportUpdateError, TimelineImportUpdateFollowUp,
 };
@@ -143,6 +153,20 @@ pub(crate) enum DatabaseOperation {
     DeleteTimelineImport,
     ListTimelineImports,
     IsTenantImportingTimeline,
+    // Brickstore Hadron
+    UpsertSafeKeeperNode,
+    LoadSafeKeepersAndEndpoints,
+    EnsureHadronEndpointTransaction,
+    DeleteHadronEndpoint,
+    GetHadronEndpointInfo,
+    FetchComputeSpec,
+    GetTenandIdByEndpointId,
+    GetTenantShardsByEndpointId,
+    GetComputeNamesByTenantId,
+    GetOrCreateHadronTimelineSafekeeper,
+    FetchPageServerAndSafeKeeperConnections,
+    DeleteHadronTimeline,
+    ListSafekeeperTimelines,
 }
 
 #[must_use]
@@ -2135,6 +2159,127 @@ impl Persistence {
         })
         .await
     }
+
+    ////////////////////////////////////////////////////////////////
+    //////////////////////// Hadron methods ////////////////////////
+    //////////////////////// (Brickstore) //////////////////////////
+    ////////////////////////////////////////////////////////////////
+
+    /// Upsert a SafeKeeper node.
+    pub(crate) async fn upsert_sk_node(&self, sk_node: &SafeKeeperNode) -> DatabaseResult<()> {
+        let sk_row = sk_node.to_database_row();
+        self.with_measured_conn(DatabaseOperation::UpsertSafeKeeperNode, move |conn| {
+            // Incantation to make the borrow checker happy
+            let sk_row_clone = sk_row.clone();
+            Box::pin(async move { execute_sk_upsert(conn, sk_row_clone).await })
+        })
+        .await
+    }
+
+    /// Load all Safe Keeper nodes and their scheduled endpoints from the database. This method is called at startup to
+    /// populate the SafeKeeperScheduler.
+    pub(crate) async fn load_safekeeper_scheduling_data(
+        &self,
+    ) -> DatabaseResult<HashMap<NodeId, SafeKeeperNode>> {
+        let sk_nodes: HashMap<NodeId, SafeKeeperNode> = self
+            .with_measured_conn(
+                DatabaseOperation::LoadSafeKeepersAndEndpoints,
+                move |conn| {
+                    // Retrieve all Safe Keeper nodes from the hadron_safekeepers table, and all timelines (grouped by
+                    // safe keeper IDs) from the hadron_timeline_safekeepers table.
+                    Box::pin(async move { scan_safekeepers_and_scheduled_timelines(conn).await })
+                },
+            )
+            .await?;
+
+        tracing::info!(
+            "load_safekeepers_and_endpoints: loaded {} safekeepers",
+            sk_nodes.len()
+        );
+
+        Ok(sk_nodes)
+    }
+
+    pub(crate) async fn get_or_assign_safekeepers_to_timeline(
+        &self,
+        timeline_id: TimelineId,
+        safekeepers: Vec<NodeId>,
+    ) -> DatabaseResult<Vec<NodeId>> {
+        self.with_measured_conn(
+            DatabaseOperation::GetOrCreateHadronTimelineSafekeeper,
+            move |conn| {
+                let safekeepers_clone = safekeepers.clone();
+                Box::pin(async move {
+                    idempotently_persist_or_get_existing_timeline_safekeepers(
+                        conn,
+                        timeline_id,
+                        &safekeepers_clone,
+                    )
+                    .await
+                })
+            },
+        )
+        .await
+    }
+
+    pub(crate) async fn delete_hadron_timeline_safekeepers(
+        &self,
+        timeline_id: TimelineId,
+    ) -> DatabaseResult<()> {
+        self.with_measured_conn(DatabaseOperation::DeleteHadronTimeline, move |conn| {
+            Box::pin(async move {
+                delete_timeline_safekeepers(conn, timeline_id).await?;
+                Ok(())
+            })
+        })
+        .await
+    }
+
+    pub(crate) async fn get_pageserver_and_safekeepers(
+        &self,
+        tenant_id: TenantId,
+        timeline_id: TimelineId,
+    ) -> DatabaseResult<PageserverAndSafekeeperConnectionInfo> {
+        self.with_measured_conn(
+            DatabaseOperation::FetchPageServerAndSafeKeeperConnections,
+            move |conn| {
+                Box::pin(async move {
+                    get_pageserver_and_safekeeper_connection_info(conn, tenant_id, timeline_id)
+                        .await
+                })
+            },
+        )
+        .await
+    }
+
+    pub(crate) async fn list_hadron_safekeepers(&self) -> DatabaseResult<Vec<HadronSafekeeperRow>> {
+        let safekeepers: Vec<HadronSafekeeperRow> = self
+            .with_measured_conn(DatabaseOperation::ListNodes, move |conn| {
+                Box::pin(async move {
+                    Ok(crate::schema::hadron_safekeepers::table
+                        .load::<HadronSafekeeperRow>(conn)
+                        .await?)
+                })
+            })
+            .await?;
+
+        tracing::info!(
+            "list_hadron_safekeepers: loaded {} nodes",
+            safekeepers.len()
+        );
+
+        Ok(safekeepers)
+    }
+
+    pub(crate) async fn safekeeper_list_timelines(
+        &self,
+        id: i64,
+    ) -> DatabaseResult<SCSafekeeperTimelinesResponse> {
+        self.with_measured_conn(DatabaseOperation::ListSafekeeperTimelines, move |conn| {
+            Box::pin(async move { execute_safekeeper_list_timelines(conn, id).await })
+        })
+        .await
+    }
 }
 
 pub(crate) fn load_certs() -> anyhow::Result<Arc<rustls::RootCertStore>> {
@@ -2238,6 +2383,53 @@ fn establish_connection_rustls(config: &str) -> BoxFuture<ConnectionResult<Async
     fut.boxed()
 }
 
+// Hadron's implementation of establish_connection_rustls which avoids hogging the tokio executor thread during
+// CPU-intensive operations in postgres connection and session establishments.
+// Compared to the original implementation this function performs the following tasks using spawn_blocking to avoid
+// hogging the tokio executor thread:
+// 1. Parsing and decoding root certificates during rustls client config setup.
+// 2. The tokio_postgres::connect() call, which performs the TLS handshake and the postgres password authentication.
+fn establish_connection_rustls_no_hog_thread(
+    config: &str,
+) -> BoxFuture<ConnectionResult<AsyncPgConnection>> {
+    let fut = async move {
+        // We first set up the way we want rustls to work.
+        let rustls_config = tokio::task::spawn_blocking(client_config_with_root_certs)
+            .await
+            .map_err(|e| {
+                ConnectionError::BadConnection(format!(
+                    "Error in spawn_blocking client_config_with_root_certs: {e}"
+                ))
+            })
+            .and_then(|r| {
+                r.map_err(|e| {
+                    ConnectionError::BadConnection(format!(
+                        "Error in client_config_with_root_certs: {e}"
+                    ))
+                })
+            })?;
+
+        let tls = tokio_postgres_rustls::MakeRustlsConnect::new(rustls_config);
+
+        // Perform the expensive TLS handshake and SCRAM SHA calculations in a blocking task
+        let task_owned_config = config.to_owned();
+        let (client, conn) = tokio::task::spawn_blocking(move || {
+            tokio::runtime::Handle::current()
+                .block_on(async { tokio_postgres::connect(&task_owned_config, tls).await })
+        })
+        .await
+        .map_err(|e| {
+            ConnectionError::BadConnection(format!(
+                "Error in spawn_blocking tokio_postgres::connect: {e}"
+            ))
+        })
+        .and_then(|r| r.map_err(|e| ConnectionError::BadConnection(e.to_string())))?;
+
+        AsyncPgConnection::try_from_client_and_connection(client, conn).await
+    };
+    fut.boxed()
+}
+
 #[cfg_attr(test, test)]
 fn test_config_debug_censors_password() {
     let has_pw =
diff --git a/storage_controller/src/sk_node.rs b/storage_controller/src/sk_node.rs
new file mode 100644
index 0000000000..9d2e835ff4
--- /dev/null
+++ b/storage_controller/src/sk_node.rs
@@ -0,0 +1,54 @@
+use serde::Serialize;
+use std::collections::{HashMap, HashSet};
+use utils::id::{NodeId, TimelineId};
+use uuid::Uuid;
+
+use crate::hadron_queries::HadronSafekeeperRow;
+
+// In-memory representation of a Safe Keeper node.
+#[derive(Clone, Serialize)]
+pub(crate) struct SafeKeeperNode {
+    pub(crate) id: NodeId,
+    pub(crate) listen_http_addr: String,
+    pub(crate) listen_http_port: u16,
+    pub(crate) listen_pg_addr: String,
+    pub(crate) listen_pg_port: u16,
+
+    // All timelines scheduled to this SK node. Some of the timelines may be associated with
+    // a legacy "endpoint", a deprecated concept used in HCC compute CRUD APIs. The "endpoint"
+    // concept will be retired after Public Preview launch.
+    pub(crate) timelines: HashSet<TimelineId>,
+    // All legacy endpoints and their associated timelines scheduled to this SK node.
+    // Invariant: The timelines referenced in this map must be present in the `timelines` set above.
+    pub(crate) legacy_endpoints: HashMap<Uuid, TimelineId>,
+}
+
+impl SafeKeeperNode {
+    pub(crate) fn new(
+        id: NodeId,
+        listen_http_addr: String,
+        listen_http_port: u16,
+        listen_pg_addr: String,
+        listen_pg_port: u16,
+    ) -> Self {
+        Self {
+            id,
+            listen_http_addr,
+            listen_http_port,
+            listen_pg_addr,
+            listen_pg_port,
+            legacy_endpoints: HashMap::new(),
+            timelines: HashSet::new(),
+        }
+    }
+
+    pub(crate) fn to_database_row(&self) -> HadronSafekeeperRow {
+        HadronSafekeeperRow {
+            sk_node_id: self.id.0 as i64,
+            listen_http_addr: self.listen_http_addr.clone(),
+            listen_http_port: self.listen_http_port as i32,
+            listen_pg_addr: self.listen_pg_addr.clone(),
+            listen_pg_port: self.listen_pg_port as i32,
+        }
+    }
+}