Merge pull request #7357 from neondatabase/rc/proxy/2024-04-11

Proxy release 2024-04-11
Remember last written LSN when it is first requested (#7343 )
2026-08-18 03:58:19 +00:00 · 2024-04-11 11:49:45 +01:00 · 2024-04-11 07:47:45 +03:00 · 2024-04-10 13:13:48 +00:00 · 2024-04-10 10:12:55 +00:00 · 2024-04-10 08:30:09 +00:00
50 changed files with 1316 additions and 731 deletions
@@ -10,7 +10,7 @@ inputs:
    required: true
  api_host:
    desctiption: 'Neon API host'
-    default: console.stage.neon.tech
+    default: console-stage.neon.build
 outputs:
  dsn:
    description: 'Created Branch DSN (for main database)'
@@ -13,7 +13,7 @@ inputs:
    required: true
  api_host:
    desctiption: 'Neon API host'
-    default: console.stage.neon.tech
+    default: console-stage.neon.build
 runs:
  using: "composite"
@@ -13,7 +13,7 @@ inputs:
    default: 15
  api_host:
    desctiption: 'Neon API host'
-    default: console.stage.neon.tech
+    default: console-stage.neon.build
  provisioner:
    desctiption: 'k8s-pod or k8s-neonvm'
    default: 'k8s-pod'
@@ -10,7 +10,7 @@ inputs:
    required: true
  api_host:
    desctiption: 'Neon API host'
-    default: console.stage.neon.tech
+    default: console-stage.neon.build
 runs:
  using: "composite"
@@ -270,6 +270,12 @@ dependencies = [
 "critical-section",
 ]
 [[package]]
 name = "atomic-take"
 version = "1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a8ab6b55fe97976e46f91ddbed8d147d966475dc29b2032757ba47e02376fbc3"
 [[package]]
 name = "autocfg"
 version = "1.1.0"
@@ -298,7 +304,7 @@ dependencies = [
 "fastrand 2.0.0",
 "hex",
 "http 0.2.9",
- "hyper",
+ "hyper 0.14.26",
 "ring 0.17.6",
 "time",
 "tokio",
@@ -335,7 +341,7 @@ dependencies = [
 "bytes",
 "fastrand 2.0.0",
 "http 0.2.9",
- "http-body",
+ "http-body 0.4.5",
 "percent-encoding",
 "pin-project-lite",
 "tracing",
@@ -386,7 +392,7 @@ dependencies = [
 "aws-types",
 "bytes",
 "http 0.2.9",
- "http-body",
+ "http-body 0.4.5",
 "once_cell",
 "percent-encoding",
 "regex-lite",
@@ -514,7 +520,7 @@ dependencies = [
 "crc32fast",
 "hex",
 "http 0.2.9",
- "http-body",
+ "http-body 0.4.5",
 "md-5",
 "pin-project-lite",
 "sha1",
@@ -546,7 +552,7 @@ dependencies = [
 "bytes-utils",
 "futures-core",
 "http 0.2.9",
- "http-body",
+ "http-body 0.4.5",
 "once_cell",
 "percent-encoding",
 "pin-project-lite",
@@ -585,10 +591,10 @@ dependencies = [
 "aws-smithy-types",
 "bytes",
 "fastrand 2.0.0",
- "h2",
+ "h2 0.3.26",
 "http 0.2.9",
- "http-body",
+ "http-body 0.4.5",
- "hyper",
+ "hyper 0.14.26",
 "hyper-rustls",
 "once_cell",
 "pin-project-lite",
@@ -626,7 +632,7 @@ dependencies = [
 "bytes-utils",
 "futures-core",
 "http 0.2.9",
- "http-body",
+ "http-body 0.4.5",
 "itoa",
 "num-integer",
 "pin-project-lite",
@@ -675,8 +681,8 @@ dependencies = [
 "bytes",
 "futures-util",
 "http 0.2.9",
- "http-body",
+ "http-body 0.4.5",
- "hyper",
+ "hyper 0.14.26",
 "itoa",
 "matchit",
 "memchr",
@@ -691,7 +697,7 @@ dependencies = [
 "sha1",
 "sync_wrapper",
 "tokio",
- "tokio-tungstenite",
+ "tokio-tungstenite 0.20.0",
 "tower",
 "tower-layer",
 "tower-service",
@@ -707,7 +713,7 @@ dependencies = [
 "bytes",
 "futures-util",
 "http 0.2.9",
- "http-body",
+ "http-body 0.4.5",
 "mime",
 "rustversion",
 "tower-layer",
@@ -1124,7 +1130,7 @@ version = "4.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "191d9573962933b4027f932c600cd252ce27a8ad5979418fe78e43c07996f27b"
 dependencies = [
- "heck",
+ "heck 0.4.1",
 "proc-macro2",
 "quote",
 "syn 2.0.52",
@@ -1196,7 +1202,7 @@ dependencies = [
 "compute_api",
 "flate2",
 "futures",
- "hyper",
+ "hyper 0.14.26",
 "nix 0.27.1",
 "notify",
 "num_cpus",
@@ -1313,7 +1319,7 @@ dependencies = [
 "git-version",
 "hex",
 "humantime",
- "hyper",
+ "hyper 0.14.26",
 "nix 0.27.1",
 "once_cell",
 "pageserver_api",
@@ -1462,12 +1468,9 @@ dependencies = [
 [[package]]
 name = "crossbeam-utils"
-version = "0.8.15"
+version = "0.8.19"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3c063cd8cc95f5c377ed0d4b49a4b21f632396ff690e8470c29b3359b346984b"
+checksum = "248e3bacc7dc6baa3b21e405ee045c3047101a49145e7e9eca583ab4c2ca5345"
 dependencies = [
 "cfg-if",
 ]
 [[package]]
 name = "crossterm"
@@ -1840,23 +1843,12 @@ checksum = "5443807d6dff69373d433ab9ef5378ad8df50ca6298caf15de6e52e24aaf54d5"
 [[package]]
 name = "errno"
-version = "0.3.1"
+version = "0.3.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4bcfec3a70f97c962c307b2d2c56e358cf1d00b558d74262b5f929ee8cc7e73a"
+checksum = "a258e46cdc063eb8519c00b9fc845fc47bcfca4130e2f08e88665ceda8474245"
 dependencies = [
 "errno-dragonfly",
 "libc",
 "windows-sys 0.48.0",
 ]
 [[package]]
 name = "errno-dragonfly"
 version = "0.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "aa68f1b12764fab894d2755d2518754e71b4fd80ecfb822714a1206c2aab39bf"
 dependencies = [
 "cc",
 "libc",
 "windows-sys 0.52.0",
 ]
 [[package]]
@@ -2213,6 +2205,25 @@ dependencies = [
 "tracing",
 ]
 [[package]]
 name = "h2"
 version = "0.4.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "816ec7294445779408f36fe57bc5b7fc1cf59664059096c65f905c1c61f58069"
 dependencies = [
 "bytes",
 "fnv",
 "futures-core",
 "futures-sink",
 "futures-util",
 "http 1.1.0",
 "indexmap 2.0.1",
 "slab",
 "tokio",
 "tokio-util",
 "tracing",
 ]
 [[package]]
 name = "half"
 version = "1.8.2"
@@ -2294,6 +2305,12 @@ version = "0.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "95505c38b4572b2d910cecb0281560f54b440a19336cbbcb27bf6ce6adc6f5a8"
 [[package]]
 name = "heck"
 version = "0.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea"
 [[package]]
 name = "hermit-abi"
 version = "0.3.3"
@@ -2378,6 +2395,29 @@ dependencies = [
 "pin-project-lite",
 ]
 [[package]]
 name = "http-body"
 version = "1.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1cac85db508abc24a2e48553ba12a996e87244a0395ce011e62b37158745d643"
 dependencies = [
 "bytes",
 "http 1.1.0",
 ]
 [[package]]
 name = "http-body-util"
 version = "0.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "41cb79eb393015dadd30fc252023adb0b2400a0caee0fa2a077e6e21a551e840"
 dependencies = [
 "bytes",
 "futures-util",
 "http 1.1.0",
 "http-body 1.0.0",
 "pin-project-lite",
 ]
 [[package]]
 name = "http-types"
 version = "2.12.0"
@@ -2436,9 +2476,9 @@ dependencies = [
 "futures-channel",
 "futures-core",
 "futures-util",
- "h2",
+ "h2 0.3.26",
 "http 0.2.9",
- "http-body",
+ "http-body 0.4.5",
 "httparse",
 "httpdate",
 "itoa",
@@ -2450,6 +2490,26 @@ dependencies = [
 "want",
 ]
 [[package]]
 name = "hyper"
 version = "1.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "186548d73ac615b32a73aafe38fb4f56c0d340e110e5a200bcadbaf2e199263a"
 dependencies = [
 "bytes",
 "futures-channel",
 "futures-util",
 "h2 0.4.4",
 "http 1.1.0",
 "http-body 1.0.0",
 "httparse",
 "httpdate",
 "itoa",
 "pin-project-lite",
 "smallvec",
 "tokio",
 ]
 [[package]]
 name = "hyper-rustls"
 version = "0.24.0"
@@ -2457,7 +2517,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0646026eb1b3eea4cd9ba47912ea5ce9cc07713d105b1a14698f4e6433d348b7"
 dependencies = [
 "http 0.2.9",
- "hyper",
+ "hyper 0.14.26",
 "log",
 "rustls 0.21.9",
 "rustls-native-certs 0.6.2",
@@ -2471,7 +2531,7 @@ version = "0.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bbb958482e8c7be4bc3cf272a766a2b0bf1a6755e7a6ae777f017a31d11b13b1"
 dependencies = [
- "hyper",
+ "hyper 0.14.26",
 "pin-project-lite",
 "tokio",
 "tokio-io-timeout",
@@ -2484,7 +2544,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d6183ddfa99b85da61a140bea0efc93fdf56ceaa041b37d553518030827f9905"
 dependencies = [
 "bytes",
- "hyper",
+ "hyper 0.14.26",
 "native-tls",
 "tokio",
 "tokio-native-tls",
@@ -2492,15 +2552,33 @@ dependencies = [
 [[package]]
 name = "hyper-tungstenite"
-version = "0.11.1"
+version = "0.13.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7cc7dcb1ab67cd336f468a12491765672e61a3b6b148634dbfe2fe8acd3fe7d9"
+checksum = "7a343d17fe7885302ed7252767dc7bb83609a874b6ff581142241ec4b73957ad"
 dependencies = [
- "hyper",
+ "http-body-util",
 "hyper 1.2.0",
 "hyper-util",
 "pin-project-lite",
 "tokio",
- "tokio-tungstenite",
+ "tokio-tungstenite 0.21.0",
- "tungstenite",
+ "tungstenite 0.21.0",
 ]
 [[package]]
 name = "hyper-util"
 version = "0.1.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ca38ef113da30126bbff9cd1705f9273e15d45498615d138b0c20279ac7a76aa"
 dependencies = [
 "bytes",
 "futures-util",
 "http 1.1.0",
 "http-body 1.0.0",
 "hyper 1.2.0",
 "pin-project-lite",
 "socket2 0.5.5",
 "tokio",
 ]
 [[package]]
@@ -2794,6 +2872,12 @@ version = "0.3.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ef53942eb7bf7ff43a617b3e2c1c4a5ecf5944a7c1bc12d7ee39bbb15e5c1519"
 [[package]]
 name = "linux-raw-sys"
 version = "0.4.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "01cda141df6706de531b6c46c3a33ecca755538219bd484262fa09410c13539c"
 [[package]]
 name = "lock_api"
 version = "0.4.10"
@@ -2848,11 +2932,12 @@ checksum = "490cc448043f947bae3cbee9c203358d62dbee0db12107a74be5c30ccfd09771"
 [[package]]
 name = "measured"
-version = "0.0.13"
+version = "0.0.20"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f246648d027839a34b420e27c7de1165ace96e19ef894985d0a6ff89a7840a9f"
+checksum = "3cbf033874bea03565f2449572c8640ca37ec26300455faf36001f24755da452"
 dependencies = [
 "bytes",
 "crossbeam-utils",
 "hashbrown 0.14.0",
 "itoa",
 "lasso",
@@ -2865,16 +2950,27 @@ dependencies = [
 [[package]]
 name = "measured-derive"
-version = "0.0.13"
+version = "0.0.20"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "edaa5cc22d99d5d6d7d99c3b5b5f7e7f8034c22f1b5d62a1adecd2ed005d9b80"
+checksum = "be9e29b682b38f8af2a89f960455054ab1a9f5a06822f6f3500637ad9fa57def"
 dependencies = [
- "heck",
+ "heck 0.5.0",
 "proc-macro2",
 "quote",
 "syn 2.0.52",
 ]
 [[package]]
 name = "measured-process"
 version = "0.0.20"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a20849acdd04c5d6a88f565559044546904648a1842a2937cfff0b48b4ca7ef2"
 dependencies = [
 "libc",
 "measured",
 "procfs 0.16.0",
 ]
 [[package]]
 name = "memchr"
 version = "2.6.4"
@@ -2914,8 +3010,10 @@ version = "0.1.0"
 dependencies = [
 "chrono",
 "libc",
 "measured",
 "measured-process",
 "once_cell",
- "procfs",
+ "procfs 0.14.2",
 "prometheus",
 "rand 0.8.5",
 "rand_distr",
@@ -3465,12 +3563,17 @@ dependencies = [
 "camino",
 "clap",
 "git-version",
 "humantime",
 "pageserver",
 "pageserver_api",
 "postgres_ffi",
 "remote_storage",
 "serde",
 "serde_json",
 "svg_fmt",
 "tokio",
 "tokio-util",
 "toml_edit",
 "utils",
 "workspace_hack",
 ]
@@ -3506,7 +3609,7 @@ dependencies = [
 "hex-literal",
 "humantime",
 "humantime-serde",
- "hyper",
+ "hyper 0.14.26",
 "itertools",
 "leaky-bucket",
 "md5",
@@ -3525,7 +3628,7 @@ dependencies = [
 "postgres_connection",
 "postgres_ffi",
 "pq_proto",
- "procfs",
+ "procfs 0.14.2",
 "rand 0.8.5",
 "regex",
 "remote_storage",
@@ -3616,7 +3719,6 @@ dependencies = [
 "anyhow",
 "async-compression",
 "async-stream",
 "async-trait",
 "byteorder",
 "bytes",
 "chrono",
@@ -4086,6 +4188,29 @@ dependencies = [
 "rustix 0.36.16",
 ]
 [[package]]
 name = "procfs"
 version = "0.16.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "731e0d9356b0c25f16f33b5be79b1c57b562f141ebfcdb0ad8ac2c13a24293b4"
 dependencies = [
 "bitflags 2.4.1",
 "hex",
 "lazy_static",
 "procfs-core",
 "rustix 0.38.28",
 ]
 [[package]]
 name = "procfs-core"
 version = "0.16.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2d3554923a69f4ce04c4a754260c338f505ce22642d3830e049a399fc2059a29"
 dependencies = [
 "bitflags 2.4.1",
 "hex",
 ]
 [[package]]
 name = "prometheus"
 version = "0.13.3"
@@ -4098,7 +4223,7 @@ dependencies = [
 "libc",
 "memchr",
 "parking_lot 0.12.1",
- "procfs",
+ "procfs 0.14.2",
 "thiserror",
 ]
@@ -4119,7 +4244,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "119533552c9a7ffacc21e099c24a0ac8bb19c2a2a3f363de84cd9b844feab270"
 dependencies = [
 "bytes",
- "heck",
+ "heck 0.4.1",
 "itertools",
 "lazy_static",
 "log",
@@ -4163,6 +4288,7 @@ dependencies = [
 "anyhow",
 "async-compression",
 "async-trait",
 "atomic-take",
 "aws-config",
 "aws-sdk-iam",
 "aws-sigv4",
@@ -4186,9 +4312,12 @@ dependencies = [
 "hmac",
 "hostname",
 "http 1.1.0",
 "http-body-util",
 "humantime",
- "hyper",
+ "hyper 0.14.26",
 "hyper 1.2.0",
 "hyper-tungstenite",
 "hyper-util",
 "ipnet",
 "itertools",
 "lasso",
@@ -4521,7 +4650,7 @@ dependencies = [
 "futures-util",
 "http-types",
 "humantime",
- "hyper",
+ "hyper 0.14.26",
 "itertools",
 "metrics",
 "once_cell",
@@ -4551,10 +4680,10 @@ dependencies = [
 "encoding_rs",
 "futures-core",
 "futures-util",
- "h2",
+ "h2 0.3.26",
 "http 0.2.9",
- "http-body",
+ "http-body 0.4.5",
- "hyper",
+ "hyper 0.14.26",
 "hyper-rustls",
 "hyper-tls",
 "ipnet",
@@ -4612,7 +4741,7 @@ dependencies = [
 "futures",
 "getrandom 0.2.11",
 "http 0.2.9",
- "hyper",
+ "hyper 0.14.26",
 "parking_lot 0.11.2",
 "reqwest",
 "reqwest-middleware",
@@ -4699,7 +4828,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "496c1d3718081c45ba9c31fbfc07417900aa96f4070ff90dc29961836b7a9945"
 dependencies = [
 "http 0.2.9",
- "hyper",
+ "hyper 0.14.26",
 "lazy_static",
 "percent-encoding",
 "regex",
@@ -4811,6 +4940,19 @@ dependencies = [
 "windows-sys 0.48.0",
 ]
 [[package]]
 name = "rustix"
 version = "0.38.28"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "72e572a5e8ca657d7366229cdde4bd14c4eb5499a9573d4d366fe1b599daa316"
 dependencies = [
 "bitflags 2.4.1",
 "errno",
 "libc",
 "linux-raw-sys 0.4.13",
 "windows-sys 0.52.0",
 ]
 [[package]]
 name = "rustls"
 version = "0.21.9"
@@ -4991,7 +5133,7 @@ dependencies = [
 "git-version",
 "hex",
 "humantime",
- "hyper",
+ "hyper 0.14.26",
 "metrics",
 "once_cell",
 "parking_lot 0.12.1",
@@ -5476,9 +5618,9 @@ dependencies = [
 [[package]]
 name = "smallvec"
-version = "1.11.0"
+version = "1.13.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "62bb4feee49fdd9f707ef802e22365a35de4b7b299de4763d44bfea899442ff9"
+checksum = "e6ecd384b10a64542d77071bd64bd7b231f4ed5940fba55e98c3de13824cf3d7"
 [[package]]
 name = "smol_str"
@@ -5570,7 +5712,7 @@ dependencies = [
 "futures-util",
 "git-version",
 "humantime",
- "hyper",
+ "hyper 0.14.26",
 "metrics",
 "once_cell",
 "parking_lot 0.12.1",
@@ -5601,7 +5743,7 @@ dependencies = [
 "git-version",
 "hex",
 "humantime",
- "hyper",
+ "hyper 0.14.26",
 "itertools",
 "lasso",
 "measured",
@@ -5630,7 +5772,7 @@ dependencies = [
 "anyhow",
 "clap",
 "comfy-table",
- "hyper",
+ "hyper 0.14.26",
 "pageserver_api",
 "pageserver_client",
 "reqwest",
@@ -5671,7 +5813,7 @@ version = "0.24.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1e385be0d24f186b4ce2f9982191e7101bb737312ad61c1f2f984f34bcf85d59"
 dependencies = [
- "heck",
+ "heck 0.4.1",
 "proc-macro2",
 "quote",
 "rustversion",
@@ -6113,7 +6255,19 @@ dependencies = [
 "futures-util",
 "log",
 "tokio",
- "tungstenite",
+ "tungstenite 0.20.1",
 ]
 [[package]]
 name = "tokio-tungstenite"
 version = "0.21.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c83b561d025642014097b66e6c1bb422783339e0909e4429cde4749d1990bc38"
 dependencies = [
 "futures-util",
 "log",
 "tokio",
 "tungstenite 0.21.0",
 ]
 [[package]]
@@ -6180,10 +6334,10 @@ dependencies = [
 "bytes",
 "futures-core",
 "futures-util",
- "h2",
+ "h2 0.3.26",
 "http 0.2.9",
- "http-body",
+ "http-body 0.4.5",
- "hyper",
+ "hyper 0.14.26",
 "hyper-timeout",
 "percent-encoding",
 "pin-project",
@@ -6369,7 +6523,7 @@ dependencies = [
 name = "tracing-utils"
 version = "0.1.0"
 dependencies = [
- "hyper",
+ "hyper 0.14.26",
 "opentelemetry",
 "opentelemetry-otlp",
 "opentelemetry-semantic-conventions",
@@ -6406,6 +6560,25 @@ dependencies = [
 "utf-8",
 ]
 [[package]]
 name = "tungstenite"
 version = "0.21.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9ef1a641ea34f399a848dea702823bbecfb4c486f911735368f1f137cb8257e1"
 dependencies = [
 "byteorder",
 "bytes",
 "data-encoding",
 "http 1.1.0",
 "httparse",
 "log",
 "rand 0.8.5",
 "sha1",
 "thiserror",
 "url",
 "utf-8",
 ]
 [[package]]
 name = "twox-hash"
 version = "1.6.3"
@@ -6570,7 +6743,8 @@ dependencies = [
 "heapless",
 "hex",
 "hex-literal",
- "hyper",
+ "humantime",
 "hyper 0.14.26",
 "jsonwebtoken",
 "leaky-bucket",
 "metrics",
@@ -6930,6 +7104,15 @@ dependencies = [
 "windows-targets 0.48.0",
 ]
 [[package]]
 name = "windows-sys"
 version = "0.52.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d"
 dependencies = [
 "windows-targets 0.52.4",
 ]
 [[package]]
 name = "windows-targets"
 version = "0.42.2"
@@ -6960,6 +7143,21 @@ dependencies = [
 "windows_x86_64_msvc 0.48.0",
 ]
 [[package]]
 name = "windows-targets"
 version = "0.52.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7dd37b7e5ab9018759f893a1952c9420d060016fc19a472b4bb20d1bdd694d1b"
 dependencies = [
 "windows_aarch64_gnullvm 0.52.4",
 "windows_aarch64_msvc 0.52.4",
 "windows_i686_gnu 0.52.4",
 "windows_i686_msvc 0.52.4",
 "windows_x86_64_gnu 0.52.4",
 "windows_x86_64_gnullvm 0.52.4",
 "windows_x86_64_msvc 0.52.4",
 ]
 [[package]]
 name = "windows_aarch64_gnullvm"
 version = "0.42.2"
@@ -6972,6 +7170,12 @@ version = "0.48.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "91ae572e1b79dba883e0d315474df7305d12f569b400fcf90581b06062f7e1bc"
 [[package]]
 name = "windows_aarch64_gnullvm"
 version = "0.52.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bcf46cf4c365c6f2d1cc93ce535f2c8b244591df96ceee75d8e83deb70a9cac9"
 [[package]]
 name = "windows_aarch64_msvc"
 version = "0.42.2"
@@ -6984,6 +7188,12 @@ version = "0.48.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b2ef27e0d7bdfcfc7b868b317c1d32c641a6fe4629c171b8928c7b08d98d7cf3"
 [[package]]
 name = "windows_aarch64_msvc"
 version = "0.52.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "da9f259dd3bcf6990b55bffd094c4f7235817ba4ceebde8e6d11cd0c5633b675"
 [[package]]
 name = "windows_i686_gnu"
 version = "0.42.2"
@@ -6996,6 +7206,12 @@ version = "0.48.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "622a1962a7db830d6fd0a69683c80a18fda201879f0f447f065a3b7467daa241"
 [[package]]
 name = "windows_i686_gnu"
 version = "0.52.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b474d8268f99e0995f25b9f095bc7434632601028cf86590aea5c8a5cb7801d3"
 [[package]]
 name = "windows_i686_msvc"
 version = "0.42.2"
@@ -7008,6 +7224,12 @@ version = "0.48.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4542c6e364ce21bf45d69fdd2a8e455fa38d316158cfd43b3ac1c5b1b19f8e00"
 [[package]]
 name = "windows_i686_msvc"
 version = "0.52.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1515e9a29e5bed743cb4415a9ecf5dfca648ce85ee42e15873c3cd8610ff8e02"
 [[package]]
 name = "windows_x86_64_gnu"
 version = "0.42.2"
@@ -7020,6 +7242,12 @@ version = "0.48.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ca2b8a661f7628cbd23440e50b05d705db3686f894fc9580820623656af974b1"
 [[package]]
 name = "windows_x86_64_gnu"
 version = "0.52.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5eee091590e89cc02ad514ffe3ead9eb6b660aedca2183455434b93546371a03"
 [[package]]
 name = "windows_x86_64_gnullvm"
 version = "0.42.2"
@@ -7032,6 +7260,12 @@ version = "0.48.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7896dbc1f41e08872e9d5e8f8baa8fdd2677f29468c4e156210174edc7f7b953"
 [[package]]
 name = "windows_x86_64_gnullvm"
 version = "0.52.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "77ca79f2451b49fa9e2af39f0747fe999fcda4f5e241b2898624dca97a1f2177"
 [[package]]
 name = "windows_x86_64_msvc"
 version = "0.42.2"
@@ -7044,6 +7278,12 @@ version = "0.48.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1a515f5799fe4961cb532f983ce2b23082366b898e52ffbce459c86f67c8378a"
 [[package]]
 name = "windows_x86_64_msvc"
 version = "0.52.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "32b752e52a2da0ddfbdbcc6fceadfeede4c939ed16d13e648833a61dfb611ed8"
 [[package]]
 name = "winnow"
 version = "0.4.6"
@@ -7092,11 +7332,10 @@ dependencies = [
 "futures-sink",
 "futures-util",
 "getrandom 0.2.11",
 "hashbrown 0.13.2",
 "hashbrown 0.14.0",
 "hex",
 "hmac",
- "hyper",
+ "hyper 0.14.26",
 "indexmap 1.9.3",
 "itertools",
 "libc",
@@ -7134,7 +7373,6 @@ dependencies = [
 "tower",
 "tracing",
 "tracing-core",
 "tungstenite",
 "url",
 "uuid",
 "zeroize",
@@ -44,6 +44,7 @@ license = "Apache-2.0"
 anyhow = { version = "1.0", features = ["backtrace"] }
 arc-swap = "1.6"
 async-compression = { version = "0.4.0", features = ["tokio", "gzip", "zstd"] }
 atomic-take = "1.1.0"
 azure_core = "0.18"
 azure_identity = "0.18"
 azure_storage = "0.18"
@@ -97,7 +98,7 @@ http-types = { version = "2", default-features = false }
 humantime = "2.1"
 humantime-serde = "1.1.1"
 hyper = "0.14"
-hyper-tungstenite = "0.11"
+hyper-tungstenite = "0.13.0"
 inotify = "0.10.2"
 ipnet = "2.9.0"
 itertools = "0.10"
@@ -106,7 +107,8 @@ lasso = "0.7"
 leaky-bucket = "1.0.1"
 libc = "0.2"
 md5 = "0.7.0"
-measured = { version = "0.0.13", features=["default", "lasso"] }
+measured = { version = "0.0.20", features=["lasso"] }
 measured-process = { version = "0.0.20" }
 memoffset = "0.8"
 native-tls = "0.2"
 nix = { version = "0.27", features = ["fs", "process", "socket", "signal", "poll"] }
@@ -6,8 +6,8 @@ use std::path::Path;
 use anyhow::Result;
 use crate::pg_helpers::escape_conf_value;
-use crate::pg_helpers::PgOptionsSerialize;
+use crate::pg_helpers::{GenericOptionExt, PgOptionsSerialize};
-use compute_api::spec::{ComputeMode, ComputeSpec};
+use compute_api::spec::{ComputeMode, ComputeSpec, GenericOption};
 /// Check that `line` is inside a text file and put it there if it is not.
 /// Create file if it doesn't exist.
@@ -92,6 +92,27 @@ pub fn write_postgres_conf(
        }
    }
    if cfg!(target_os = "linux") {
        // Check /proc/sys/vm/overcommit_memory -- if it equals 2 (i.e. linux memory overcommit is
        // disabled), then the control plane has enabled swap and we should set
        // dynamic_shared_memory_type = 'mmap'.
        //
        // This is (maybe?) temporary - for more, see https://github.com/neondatabase/cloud/issues/12047.
        let overcommit_memory_contents = std::fs::read_to_string("/proc/sys/vm/overcommit_memory")
            // ignore any errors - they may be expected to occur under certain situations (e.g. when
            // not running in Linux).
            .unwrap_or_else(|_| String::new());
        if overcommit_memory_contents.trim() == "2" {
            let opt = GenericOption {
                name: "dynamic_shared_memory_type".to_owned(),
                value: Some("mmap".to_owned()),
                vartype: "enum".to_owned(),
            };
            write!(file, "{}", opt.to_pg_setting())?;
        }
    }
    // If there are any extra options in the 'settings' field, append those
    if spec.cluster.settings.is_some() {
        writeln!(file, "# Managed by compute_ctl: begin")?;
@@ -44,7 +44,7 @@ pub fn escape_conf_value(s: &str) -> String {
    format!("'{}'", res)
 }
-trait GenericOptionExt {
+pub trait GenericOptionExt {
    fn to_pg_option(&self) -> String;
    fn to_pg_setting(&self) -> String;
 }
@@ -86,7 +86,10 @@ where
        .stdout(process_log_file)
        .stderr(same_file_for_stderr)
        .args(args);
-    let filled_cmd = fill_remote_storage_secrets_vars(fill_rust_env_vars(background_command));
+
    let filled_cmd = fill_env_vars_prefixed_neon(fill_remote_storage_secrets_vars(
        fill_rust_env_vars(background_command),
    ));
    filled_cmd.envs(envs);
    let pid_file_to_check = match &initial_pid_file {
@@ -268,6 +271,15 @@ fn fill_remote_storage_secrets_vars(mut cmd: &mut Command) -> &mut Command {
    cmd
 }
 fn fill_env_vars_prefixed_neon(mut cmd: &mut Command) -> &mut Command {
    for (var, val) in std::env::vars() {
        if var.starts_with("NEON_PAGESERVER_") {
            cmd = cmd.env(var, val);
        }
    }
    cmd
 }
 /// Add a `pre_exec` to the cmd that, inbetween fork() and exec(),
 /// 1. Claims a pidfile with a fcntl lock on it and
 /// 2. Sets up the pidfile's file descriptor so that it (and the lock)
@@ -10,11 +10,13 @@ libc.workspace = true
 once_cell.workspace = true
 chrono.workspace = true
 twox-hash.workspace = true
 measured.workspace = true
 workspace_hack.workspace = true
 [target.'cfg(target_os = "linux")'.dependencies]
 procfs.workspace = true
 measured-process.workspace = true
 [dev-dependencies]
 rand = "0.8"
@@ -4,6 +4,17 @@
 //! a default registry.
 #![deny(clippy::undocumented_unsafe_blocks)]
 use measured::{
    label::{LabelGroupVisitor, LabelName, NoLabels},
    metric::{
        counter::CounterState,
        gauge::GaugeState,
        group::{Encoding, MetricValue},
        name::{MetricName, MetricNameEncoder},
        MetricEncoding, MetricFamilyEncoding,
    },
    FixedCardinalityLabel, LabelGroup, MetricGroup,
 };
 use once_cell::sync::Lazy;
 use prometheus::core::{
    Atomic, AtomicU64, Collector, GenericCounter, GenericCounterVec, GenericGauge, GenericGaugeVec,
@@ -11,6 +22,7 @@ use prometheus::core::{
 pub use prometheus::opts;
 pub use prometheus::register;
 pub use prometheus::Error;
 use prometheus::Registry;
 pub use prometheus::{core, default_registry, proto};
 pub use prometheus::{exponential_buckets, linear_buckets};
 pub use prometheus::{register_counter_vec, Counter, CounterVec};
@@ -23,7 +35,6 @@ pub use prometheus::{register_int_counter_vec, IntCounterVec};
 pub use prometheus::{register_int_gauge, IntGauge};
 pub use prometheus::{register_int_gauge_vec, IntGaugeVec};
 pub use prometheus::{Encoder, TextEncoder};
 use prometheus::{Registry, Result};
 pub mod launch_timestamp;
 mod wrappers;
@@ -59,7 +70,7 @@ static INTERNAL_REGISTRY: Lazy<Registry> = Lazy::new(Registry::new);
 /// Register a collector in the internal registry. MUST be called before the first call to `gather()`.
 /// Otherwise, we can have a deadlock in the `gather()` call, trying to register a new collector
 /// while holding the lock.
-pub fn register_internal(c: Box<dyn Collector>) -> Result<()> {
+pub fn register_internal(c: Box<dyn Collector>) -> prometheus::Result<()> {
    INTERNAL_REGISTRY.register(c)
 }
@@ -96,6 +107,127 @@ pub const DISK_WRITE_SECONDS_BUCKETS: &[f64] = &[
    0.000_050, 0.000_100, 0.000_500, 0.001, 0.003, 0.005, 0.01, 0.05, 0.1, 0.3, 0.5,
 ];
 pub struct BuildInfo {
    pub revision: &'static str,
    pub build_tag: &'static str,
 }
 // todo: allow label group without the set
 impl LabelGroup for BuildInfo {
    fn visit_values(&self, v: &mut impl LabelGroupVisitor) {
        const REVISION: &LabelName = LabelName::from_str("revision");
        v.write_value(REVISION, &self.revision);
        const BUILD_TAG: &LabelName = LabelName::from_str("build_tag");
        v.write_value(BUILD_TAG, &self.build_tag);
    }
 }
 impl<T: Encoding> MetricFamilyEncoding<T> for BuildInfo
 where
    GaugeState: MetricEncoding<T>,
 {
    fn collect_family_into(
        &self,
        name: impl measured::metric::name::MetricNameEncoder,
        enc: &mut T,
    ) -> Result<(), T::Err> {
        enc.write_help(&name, "Build/version information")?;
        GaugeState::write_type(&name, enc)?;
        GaugeState {
            count: std::sync::atomic::AtomicI64::new(1),
        }
        .collect_into(&(), self, name, enc)
    }
 }
 #[derive(MetricGroup)]
 #[metric(new(build_info: BuildInfo))]
 pub struct NeonMetrics {
    #[cfg(target_os = "linux")]
    #[metric(namespace = "process")]
    #[metric(init = measured_process::ProcessCollector::for_self())]
    process: measured_process::ProcessCollector,
    #[metric(namespace = "libmetrics")]
    #[metric(init = LibMetrics::new(build_info))]
    libmetrics: LibMetrics,
 }
 #[derive(MetricGroup)]
 #[metric(new(build_info: BuildInfo))]
 pub struct LibMetrics {
    #[metric(init = build_info)]
    build_info: BuildInfo,
    #[metric(flatten)]
    rusage: Rusage,
    serve_count: CollectionCounter,
 }
 fn write_gauge<Enc: Encoding>(
    x: i64,
    labels: impl LabelGroup,
    name: impl MetricNameEncoder,
    enc: &mut Enc,
 ) -> Result<(), Enc::Err> {
    enc.write_metric_value(name, labels, MetricValue::Int(x))
 }
 #[derive(Default)]
 struct Rusage;
 #[derive(FixedCardinalityLabel, Clone, Copy)]
 #[label(singleton = "io_operation")]
 enum IoOp {
    Read,
    Write,
 }
 impl<T: Encoding> MetricGroup<T> for Rusage
 where
    GaugeState: MetricEncoding<T>,
 {
    fn collect_group_into(&self, enc: &mut T) -> Result<(), T::Err> {
        const DISK_IO: &MetricName = MetricName::from_str("disk_io_bytes_total");
        const MAXRSS: &MetricName = MetricName::from_str("maxrss_kb");
        let ru = get_rusage_stats();
        enc.write_help(
            DISK_IO,
            "Bytes written and read from disk, grouped by the operation (read|write)",
        )?;
        GaugeState::write_type(DISK_IO, enc)?;
        write_gauge(ru.ru_inblock * BYTES_IN_BLOCK, IoOp::Read, DISK_IO, enc)?;
        write_gauge(ru.ru_oublock * BYTES_IN_BLOCK, IoOp::Write, DISK_IO, enc)?;
        enc.write_help(MAXRSS, "Memory usage (Maximum Resident Set Size)")?;
        GaugeState::write_type(MAXRSS, enc)?;
        write_gauge(ru.ru_maxrss, IoOp::Read, MAXRSS, enc)?;
        Ok(())
    }
 }
 #[derive(Default)]
 struct CollectionCounter(CounterState);
 impl<T: Encoding> MetricFamilyEncoding<T> for CollectionCounter
 where
    CounterState: MetricEncoding<T>,
 {
    fn collect_family_into(
        &self,
        name: impl measured::metric::name::MetricNameEncoder,
        enc: &mut T,
    ) -> Result<(), T::Err> {
        self.0.inc();
        enc.write_help(&name, "Number of metric requests made")?;
        self.0.collect_into(&(), NoLabels, name, enc)
    }
 }
 pub fn set_build_info_metric(revision: &str, build_tag: &str) {
    let metric = register_int_gauge_vec!(
        "libmetrics_build_info",
@@ -105,6 +237,7 @@ pub fn set_build_info_metric(revision: &str, build_tag: &str) {
    .expect("Failed to register build info metric");
    metric.with_label_values(&[revision, build_tag]).set(1);
 }
 const BYTES_IN_BLOCK: i64 = 512;
 // Records I/O stats in a "cross-platform" way.
 // Compiles both on macOS and Linux, but current macOS implementation always returns 0 as values for I/O stats.
@@ -117,7 +250,6 @@ pub fn set_build_info_metric(revision: &str, build_tag: &str) {
 fn update_rusage_metrics() {
    let rusage_stats = get_rusage_stats();
    const BYTES_IN_BLOCK: i64 = 512;
    DISK_IO_BYTES
        .with_label_values(&["read"])
        .set(rusage_stats.ru_inblock * BYTES_IN_BLOCK);
@@ -151,6 +283,7 @@ macro_rules! register_int_counter_pair_vec {
        }
    }};
 }
 /// Create an [`IntCounterPair`] and registers to default registry.
 #[macro_export(local_inner_macros)]
 macro_rules! register_int_counter_pair {
@@ -188,7 +321,10 @@ impl<P: Atomic> GenericCounterPairVec<P> {
    ///
    /// An error is returned if the number of label values is not the same as the
    /// number of VariableLabels in Desc.
-    pub fn get_metric_with_label_values(&self, vals: &[&str]) -> Result<GenericCounterPair<P>> {
+    pub fn get_metric_with_label_values(
        &self,
        vals: &[&str],
    ) -> prometheus::Result<GenericCounterPair<P>> {
        Ok(GenericCounterPair {
            inc: self.inc.get_metric_with_label_values(vals)?,
            dec: self.dec.get_metric_with_label_values(vals)?,
@@ -201,7 +337,7 @@ impl<P: Atomic> GenericCounterPairVec<P> {
        self.get_metric_with_label_values(vals).unwrap()
    }
-    pub fn remove_label_values(&self, res: &mut [Result<()>; 2], vals: &[&str]) {
+    pub fn remove_label_values(&self, res: &mut [prometheus::Result<()>; 2], vals: &[&str]) {
        res[0] = self.inc.remove_label_values(vals);
        res[1] = self.dec.remove_label_values(vals);
    }
@@ -20,6 +20,7 @@ use utils::{
    history_buffer::HistoryBufferWithDropCounter,
    id::{NodeId, TenantId, TimelineId},
    lsn::Lsn,
    serde_system_time,
 };
 use crate::controller_api::PlacementPolicy;
@@ -758,11 +759,7 @@ pub struct WalRedoManagerStatus {
 #[derive(Default, Debug, Serialize, Deserialize, Clone)]
 pub struct SecondaryProgress {
    /// The remote storage LastModified time of the heatmap object we last downloaded.
-    #[serde(
+    pub heatmap_mtime: Option<serde_system_time::SystemTime>,
        serialize_with = "opt_ser_rfc3339_millis",
        deserialize_with = "opt_deser_rfc3339_millis"
    )]
    pub heatmap_mtime: Option<SystemTime>,
    /// The number of layers currently on-disk
    pub layers_downloaded: usize,
@@ -775,29 +772,6 @@ pub struct SecondaryProgress {
    pub bytes_total: u64,
 }
 fn opt_ser_rfc3339_millis<S: serde::Serializer>(
    ts: &Option<SystemTime>,
    serializer: S,
 ) -> Result<S::Ok, S::Error> {
    match ts {
        Some(ts) => serializer.collect_str(&humantime::format_rfc3339_millis(*ts)),
        None => serializer.serialize_none(),
    }
 }
 fn opt_deser_rfc3339_millis<'de, D>(deserializer: D) -> Result<Option<SystemTime>, D::Error>
 where
    D: serde::de::Deserializer<'de>,
 {
    let s: Option<String> = serde::de::Deserialize::deserialize(deserializer)?;
    match s {
        None => Ok(None),
        Some(s) => humantime::parse_rfc3339(&s)
            .map_err(serde::de::Error::custom)
            .map(Some),
    }
 }
 pub mod virtual_file {
    #[derive(
        Copy,
@@ -1,4 +1,4 @@
-use std::time::SystemTime;
+use utils::serde_system_time::SystemTime;
 /// Pageserver current utilization and scoring for how good candidate the pageserver would be for
 /// the next tenant.
@@ -21,28 +21,9 @@ pub struct PageserverUtilization {
    /// When was this snapshot captured, pageserver local time.
    ///
    /// Use millis to give confidence that the value is regenerated often enough.
    #[serde(
        serialize_with = "ser_rfc3339_millis",
        deserialize_with = "deser_rfc3339_millis"
    )]
    pub captured_at: SystemTime,
 }
 fn ser_rfc3339_millis<S: serde::Serializer>(
    ts: &SystemTime,
    serializer: S,
 ) -> Result<S::Ok, S::Error> {
    serializer.collect_str(&humantime::format_rfc3339_millis(*ts))
 }
 fn deser_rfc3339_millis<'de, D>(deserializer: D) -> Result<SystemTime, D::Error>
 where
    D: serde::de::Deserializer<'de>,
 {
    let s: String = serde::de::Deserialize::deserialize(deserializer)?;
    humantime::parse_rfc3339(&s).map_err(serde::de::Error::custom)
 }
 /// openapi knows only `format: int64`, so avoid outputting a non-parseable value by generated clients.
 ///
 /// Instead of newtype, use this because a newtype would get require handling deserializing values
@@ -69,7 +50,9 @@ mod tests {
            disk_usage_bytes: u64::MAX,
            free_space_bytes: 0,
            utilization_score: u64::MAX,
-            captured_at: SystemTime::UNIX_EPOCH + Duration::from_secs(1708509779),
+            captured_at: SystemTime(
                std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(1708509779),
            ),
        };
        let s = serde_json::to_string(&doc).unwrap();
@@ -22,6 +22,7 @@ camino.workspace = true
 chrono.workspace = true
 heapless.workspace = true
 hex = { workspace = true, features = ["serde"] }
 humantime.workspace = true
 hyper = { workspace = true, features = ["full"] }
 fail.workspace = true
 futures = { workspace = true}
@@ -0,0 +1,21 @@
 //! Wrapper around `std::env::var` for parsing environment variables.
 use std::{fmt::Display, str::FromStr};
 pub fn var<V, E>(varname: &str) -> Option<V>
 where
    V: FromStr<Err = E>,
    E: Display,
 {
    match std::env::var(varname) {
        Ok(s) => Some(
            s.parse()
                .map_err(|e| format!("failed to parse env var {varname}: {e:#}"))
                .unwrap(),
        ),
        Err(std::env::VarError::NotPresent) => None,
        Err(std::env::VarError::NotUnicode(_)) => {
            panic!("env var {varname} is not unicode")
        }
    }
 }
@@ -63,6 +63,7 @@ pub mod measured_stream;
 pub mod serde_percent;
 pub mod serde_regex;
 pub mod serde_system_time;
 pub mod pageserver_feedback;
@@ -89,6 +90,8 @@ pub mod yielding_loop;
 pub mod zstd;
 pub mod env;
 /// This is a shortcut to embed git sha into binaries and avoid copying the same build script to all packages
 ///
 /// we have several cases:
@@ -0,0 +1,55 @@
 //! A `serde::{Deserialize,Serialize}` type for SystemTime with RFC3339 format and millisecond precision.
 #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, serde::Serialize, serde::Deserialize)]
 #[serde(transparent)]
 pub struct SystemTime(
    #[serde(
        deserialize_with = "deser_rfc3339_millis",
        serialize_with = "ser_rfc3339_millis"
    )]
    pub std::time::SystemTime,
 );
 fn ser_rfc3339_millis<S: serde::ser::Serializer>(
    ts: &std::time::SystemTime,
    serializer: S,
 ) -> Result<S::Ok, S::Error> {
    serializer.collect_str(&humantime::format_rfc3339_millis(*ts))
 }
 fn deser_rfc3339_millis<'de, D>(deserializer: D) -> Result<std::time::SystemTime, D::Error>
 where
    D: serde::de::Deserializer<'de>,
 {
    let s: String = serde::de::Deserialize::deserialize(deserializer)?;
    humantime::parse_rfc3339(&s).map_err(serde::de::Error::custom)
 }
 #[cfg(test)]
 mod tests {
    use super::*;
    /// Helper function to make a SystemTime have millisecond precision by truncating additional nanoseconds.
    fn to_millisecond_precision(time: SystemTime) -> SystemTime {
        match time.0.duration_since(std::time::SystemTime::UNIX_EPOCH) {
            Ok(duration) => {
                let total_millis = duration.as_secs() * 1_000 + u64::from(duration.subsec_millis());
                SystemTime(
                    std::time::SystemTime::UNIX_EPOCH
                        + std::time::Duration::from_millis(total_millis),
                )
            }
            Err(_) => time,
        }
    }
    #[test]
    fn test_serialize_deserialize() {
        let input = SystemTime(std::time::SystemTime::now());
        let expected_serialized = format!("\"{}\"", humantime::format_rfc3339_millis(input.0));
        let serialized = serde_json::to_string(&input).unwrap();
        assert_eq!(expected_serialized, serialized);
        let deserialized: SystemTime = serde_json::from_str(&expected_serialized).unwrap();
        assert_eq!(to_millisecond_precision(input), deserialized);
    }
 }
@@ -11,7 +11,6 @@ default = []
 anyhow.workspace = true
 async-compression.workspace = true
 async-stream.workspace = true
 async-trait.workspace = true
 byteorder.workspace = true
 bytes.workspace = true
 chrono = { workspace = true, features = ["serde"] }
@@ -180,7 +180,7 @@ where
                match top.deref_mut() {
                    LazyLoadLayer::Unloaded(ref mut l) => {
                        let fut = l.load_keys(this.ctx);
-                        this.load_future.set(Some(fut));
+                        this.load_future.set(Some(Box::pin(fut)));
                        continue;
                    }
                    LazyLoadLayer::Loaded(ref mut entries) => {
@@ -3,7 +3,6 @@
 //!
 //! All the heavy lifting is done by the create_image and create_delta
 //! functions that the implementor provides.
 use async_trait::async_trait;
 use futures::Future;
 use pageserver_api::{key::Key, keyspace::key_range_size};
 use std::ops::Range;
@@ -141,18 +140,16 @@ pub trait CompactionLayer<K: CompactionKey + ?Sized> {
    fn is_delta(&self) -> bool;
 }
 #[async_trait]
 pub trait CompactionDeltaLayer<E: CompactionJobExecutor + ?Sized>: CompactionLayer<E::Key> {
    type DeltaEntry<'a>: CompactionDeltaEntry<'a, E::Key>
    where
        Self: 'a;
    /// Return all keys in this delta layer.
-    async fn load_keys<'a>(
+    fn load_keys<'a>(
        &self,
        ctx: &E::RequestContext,
-    ) -> anyhow::Result<Vec<Self::DeltaEntry<'_>>>;
+    ) -> impl Future<Output = anyhow::Result<Vec<Self::DeltaEntry<'_>>>> + Send;
 }
 pub trait CompactionImageLayer<E: CompactionJobExecutor + ?Sized>: CompactionLayer<E::Key> {}
@@ -2,7 +2,6 @@ mod draw;
 use draw::{LayerTraceEvent, LayerTraceFile, LayerTraceOp};
 use async_trait::async_trait;
 use futures::StreamExt;
 use rand::Rng;
 use tracing::info;
@@ -139,7 +138,6 @@ impl interface::CompactionLayer<Key> for Arc<MockDeltaLayer> {
    }
 }
 #[async_trait]
 impl interface::CompactionDeltaLayer<MockTimeline> for Arc<MockDeltaLayer> {
    type DeltaEntry<'a> = MockRecord;
@@ -12,9 +12,14 @@ bytes.workspace = true
 camino.workspace = true
 clap = { workspace = true, features = ["string"] }
 git-version.workspace = true
 humantime.workspace = true
 pageserver = { path = ".." }
 pageserver_api.workspace = true
 remote_storage = { path = "../../libs/remote_storage" }
 postgres_ffi.workspace = true
 tokio.workspace = true
 tokio-util.workspace = true
 toml_edit.workspace = true
 utils.workspace = true
 svg_fmt.workspace = true
 workspace_hack.workspace = true
@@ -9,6 +9,11 @@ mod index_part;
 mod layer_map_analyzer;
 mod layers;
 use std::{
    str::FromStr,
    time::{Duration, SystemTime},
 };
 use camino::{Utf8Path, Utf8PathBuf};
 use clap::{Parser, Subcommand};
 use index_part::IndexPartCmd;
@@ -20,8 +25,16 @@ use pageserver::{
    tenant::{dump_layerfile_from_path, metadata::TimelineMetadata},
    virtual_file,
 };
 use pageserver_api::shard::TenantShardId;
 use postgres_ffi::ControlFileData;
-use utils::{lsn::Lsn, project_git_version};
+use remote_storage::{RemotePath, RemoteStorageConfig};
 use tokio_util::sync::CancellationToken;
 use utils::{
    id::TimelineId,
    logging::{self, LogFormat, TracingErrorLayerEnablement},
    lsn::Lsn,
    project_git_version,
 };
 project_git_version!(GIT_VERSION);
@@ -43,6 +56,7 @@ enum Commands {
    #[command(subcommand)]
    IndexPart(IndexPartCmd),
    PrintLayerFile(PrintLayerFileCmd),
    TimeTravelRemotePrefix(TimeTravelRemotePrefixCmd),
    DrawTimeline {},
    AnalyzeLayerMap(AnalyzeLayerMapCmd),
    #[command(subcommand)]
@@ -68,6 +82,26 @@ struct PrintLayerFileCmd {
    path: Utf8PathBuf,
 }
 /// Roll back the time for the specified prefix using S3 history.
 ///
 /// The command is fairly low level and powerful. Validation is only very light,
 /// so it is more powerful, and thus potentially more dangerous.
 #[derive(Parser)]
 struct TimeTravelRemotePrefixCmd {
    /// A configuration string for the remote_storage configuration.
    ///
    /// Example: `remote_storage = { bucket_name = "aws-storage-bucket-name", bucket_region = "us-east-2" }`
    config_toml_str: String,
    /// remote prefix to time travel recover. For safety reasons, we require it to contain
    /// a timeline or tenant ID in the prefix.
    prefix: String,
    /// Timestamp to travel to. Given in format like `2024-01-20T10:45:45Z`. Assumes UTC and second accuracy.
    travel_to: String,
    /// Timestamp of the start of the operation, must be after any changes we want to roll back and after.
    /// You can use a few seconds before invoking the command. Same format as `travel_to`.
    done_if_after: Option<String>,
 }
 #[derive(Parser)]
 struct AnalyzeLayerMapCmd {
    /// Pageserver data path
@@ -78,6 +112,14 @@ struct AnalyzeLayerMapCmd {
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
    logging::init(
        LogFormat::Plain,
        TracingErrorLayerEnablement::EnableWithRustLogFilter,
        logging::Output::Stdout,
    )?;
    logging::replace_panic_hook_with_tracing_panic_hook().forget();
    let cli = CliOpts::parse();
    match cli.command {
@@ -105,6 +147,42 @@ async fn main() -> anyhow::Result<()> {
                print_layerfile(&cmd.path).await?;
            }
        }
        Commands::TimeTravelRemotePrefix(cmd) => {
            let timestamp = humantime::parse_rfc3339(&cmd.travel_to)
                .map_err(|_e| anyhow::anyhow!("Invalid time for travel_to: '{}'", cmd.travel_to))?;
            let done_if_after = if let Some(done_if_after) = &cmd.done_if_after {
                humantime::parse_rfc3339(done_if_after).map_err(|_e| {
                    anyhow::anyhow!("Invalid time for done_if_after: '{}'", done_if_after)
                })?
            } else {
                const SAFETY_MARGIN: Duration = Duration::from_secs(3);
                tokio::time::sleep(SAFETY_MARGIN).await;
                // Convert to string representation and back to get rid of sub-second values
                let done_if_after = SystemTime::now();
                tokio::time::sleep(SAFETY_MARGIN).await;
                done_if_after
            };
            let timestamp = strip_subsecond(timestamp);
            let done_if_after = strip_subsecond(done_if_after);
            let Some(prefix) = validate_prefix(&cmd.prefix) else {
                println!("specified prefix '{}' failed validation", cmd.prefix);
                return Ok(());
            };
            let toml_document = toml_edit::Document::from_str(&cmd.config_toml_str)?;
            let toml_item = toml_document
                .get("remote_storage")
                .expect("need remote_storage");
            let config = RemoteStorageConfig::from_toml(toml_item)?.expect("incomplete config");
            let storage = remote_storage::GenericRemoteStorage::from_config(&config);
            let cancel = CancellationToken::new();
            storage
                .unwrap()
                .time_travel_recover(Some(&prefix), timestamp, done_if_after, &cancel)
                .await?;
        }
    };
    Ok(())
 }
@@ -185,3 +263,89 @@ fn handle_metadata(
    Ok(())
 }
 /// Ensures that the given S3 prefix is sufficiently constrained.
 /// The command is very risky already and we don't want to expose something
 /// that allows usually unintentional and quite catastrophic time travel of
 /// an entire bucket, which would be a major catastrophy and away
 /// by only one character change (similar to "rm -r /home /username/foobar").
 fn validate_prefix(prefix: &str) -> Option<RemotePath> {
    if prefix.is_empty() {
        // Empty prefix means we want to specify the *whole* bucket
        return None;
    }
    let components = prefix.split('/').collect::<Vec<_>>();
    let (last, components) = {
        let last = components.last()?;
        if last.is_empty() {
            (
                components.iter().nth_back(1)?,
                &components[..(components.len() - 1)],
            )
        } else {
            (last, &components[..])
        }
    };
    'valid: {
        if let Ok(_timeline_id) = TimelineId::from_str(last) {
            // Ends in either a tenant or timeline ID
            break 'valid;
        }
        if *last == "timelines" {
            if let Some(before_last) = components.iter().nth_back(1) {
                if let Ok(_tenant_id) = TenantShardId::from_str(before_last) {
                    // Has a valid tenant id
                    break 'valid;
                }
            }
        }
        return None;
    }
    RemotePath::from_string(prefix).ok()
 }
 fn strip_subsecond(timestamp: SystemTime) -> SystemTime {
    let ts_str = humantime::format_rfc3339_seconds(timestamp).to_string();
    humantime::parse_rfc3339(&ts_str).expect("can't parse just created timestamp")
 }
 #[cfg(test)]
 mod tests {
    use super::*;
    #[test]
    fn test_validate_prefix() {
        assert_eq!(validate_prefix(""), None);
        assert_eq!(validate_prefix("/"), None);
        #[track_caller]
        fn assert_valid(prefix: &str) {
            let remote_path = RemotePath::from_string(prefix).unwrap();
            assert_eq!(validate_prefix(prefix), Some(remote_path));
        }
        assert_valid("wal/3aa8fcc61f6d357410b7de754b1d9001/641e5342083b2235ee3deb8066819683/");
        // Path is not relative but absolute
        assert_eq!(
            validate_prefix(
                "/wal/3aa8fcc61f6d357410b7de754b1d9001/641e5342083b2235ee3deb8066819683/"
            ),
            None
        );
        assert_valid("wal/3aa8fcc61f6d357410b7de754b1d9001/");
        // Partial tenant IDs should be invalid, S3 will match all tenants with the specific ID prefix
        assert_eq!(validate_prefix("wal/3aa8fcc61f6d357410b7d"), None);
        assert_eq!(validate_prefix("wal"), None);
        assert_eq!(validate_prefix("/wal/"), None);
        assert_valid("pageserver/v1/tenants/3aa8fcc61f6d357410b7de754b1d9001");
        // Partial tenant ID
        assert_eq!(
            validate_prefix("pageserver/v1/tenants/3aa8fcc61f6d357410b"),
            None
        );
        assert_valid("pageserver/v1/tenants/3aa8fcc61f6d357410b7de754b1d9001/timelines");
        assert_valid("pageserver/v1/tenants/3aa8fcc61f6d357410b7de754b1d9001-0004/timelines");
        assert_valid("pageserver/v1/tenants/3aa8fcc61f6d357410b7de754b1d9001/timelines/");
        assert_valid("pageserver/v1/tenants/3aa8fcc61f6d357410b7de754b1d9001/timelines/641e5342083b2235ee3deb8066819683");
        assert_eq!(validate_prefix("pageserver/v1/tenants/"), None);
    }
 }
@@ -2100,6 +2100,7 @@ pub(crate) fn remove_tenant_metrics(tenant_shard_id: &TenantShardId) {
 use futures::Future;
 use pin_project_lite::pin_project;
 use std::collections::HashMap;
 use std::num::NonZeroUsize;
 use std::pin::Pin;
 use std::sync::{Arc, Mutex};
 use std::task::{Context, Poll};
@@ -2669,6 +2670,26 @@ pub(crate) mod disk_usage_based_eviction {
    pub(crate) static METRICS: Lazy<Metrics> = Lazy::new(Metrics::default);
 }
 static TOKIO_EXECUTOR_THREAD_COUNT: Lazy<UIntGaugeVec> = Lazy::new(|| {
    register_uint_gauge_vec!(
        "pageserver_tokio_executor_thread_configured_count",
        "Total number of configued tokio executor threads in the process.
         The `setup` label denotes whether we're running with multiple runtimes or a single runtime.",
        &["setup"],
    )
    .unwrap()
 });
 pub(crate) fn set_tokio_runtime_setup(setup: &str, num_threads: NonZeroUsize) {
    static SERIALIZE: std::sync::Mutex<()> = std::sync::Mutex::new(());
    let _guard = SERIALIZE.lock().unwrap();
    TOKIO_EXECUTOR_THREAD_COUNT.reset();
    TOKIO_EXECUTOR_THREAD_COUNT
        .get_metric_with_label_values(&[setup])
        .unwrap()
        .set(u64::try_from(num_threads.get()).unwrap());
 }
 pub fn preinitialize_metrics() {
    // Python tests need these and on some we do alerting.
    //
@@ -33,13 +33,14 @@
 use std::collections::HashMap;
 use std::fmt;
 use std::future::Future;
 use std::num::NonZeroUsize;
 use std::panic::AssertUnwindSafe;
 use std::str::FromStr;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
 use futures::FutureExt;
 use pageserver_api::shard::TenantShardId;
 use tokio::runtime::Runtime;
 use tokio::task::JoinHandle;
 use tokio::task_local;
 use tokio_util::sync::CancellationToken;
@@ -48,8 +49,11 @@ use tracing::{debug, error, info, warn};
 use once_cell::sync::Lazy;
 use utils::env;
 use utils::id::TimelineId;
 use crate::metrics::set_tokio_runtime_setup;
 //
 // There are four runtimes:
 //
@@ -98,52 +102,119 @@ use utils::id::TimelineId;
 // other operations, if the upload tasks e.g. get blocked on locks. It shouldn't
 // happen, but still.
 //
 pub static COMPUTE_REQUEST_RUNTIME: Lazy<Runtime> = Lazy::new(|| {
    tokio::runtime::Builder::new_multi_thread()
        .thread_name("compute request worker")
        .enable_all()
        .build()
        .expect("Failed to create compute request runtime")
 });
-pub static MGMT_REQUEST_RUNTIME: Lazy<Runtime> = Lazy::new(|| {
+pub(crate) static TOKIO_WORKER_THREADS: Lazy<NonZeroUsize> = Lazy::new(|| {
    tokio::runtime::Builder::new_multi_thread()
        .thread_name("mgmt request worker")
        .enable_all()
        .build()
        .expect("Failed to create mgmt request runtime")
 });
 pub static WALRECEIVER_RUNTIME: Lazy<Runtime> = Lazy::new(|| {
    tokio::runtime::Builder::new_multi_thread()
        .thread_name("walreceiver worker")
        .enable_all()
        .build()
        .expect("Failed to create walreceiver runtime")
 });
 pub static BACKGROUND_RUNTIME: Lazy<Runtime> = Lazy::new(|| {
    tokio::runtime::Builder::new_multi_thread()
        .thread_name("background op worker")
        // if you change the number of worker threads please change the constant below
        .enable_all()
        .build()
        .expect("Failed to create background op runtime")
 });
 pub(crate) static BACKGROUND_RUNTIME_WORKER_THREADS: Lazy<usize> = Lazy::new(|| {
    // force init and thus panics
    let _ = BACKGROUND_RUNTIME.handle();
    // replicates tokio-1.28.1::loom::sys::num_cpus which is not available publicly
    // tokio would had already panicked for parsing errors or NotUnicode
    //
    // this will be wrong if any of the runtimes gets their worker threads configured to something
    // else, but that has not been needed in a long time.
-    std::env::var("TOKIO_WORKER_THREADS")
+    NonZeroUsize::new(
-        .map(|s| s.parse::<usize>().unwrap())
+        std::env::var("TOKIO_WORKER_THREADS")
-        .unwrap_or_else(|_e| usize::max(2, num_cpus::get()))
+            .map(|s| s.parse::<usize>().unwrap())
            .unwrap_or_else(|_e| usize::max(2, num_cpus::get())),
    )
    .expect("the max() ensures that this is not zero")
 });
 enum TokioRuntimeMode {
    SingleThreaded,
    MultiThreaded { num_workers: NonZeroUsize },
 }
 impl FromStr for TokioRuntimeMode {
    type Err = String;
    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s {
            "current_thread" => Ok(TokioRuntimeMode::SingleThreaded),
            s => match s.strip_prefix("multi_thread:") {
                Some("default") => Ok(TokioRuntimeMode::MultiThreaded {
                    num_workers: *TOKIO_WORKER_THREADS,
                }),
                Some(suffix) => {
                    let num_workers = suffix.parse::<NonZeroUsize>().map_err(|e| {
                        format!(
                            "invalid number of multi-threaded runtime workers ({suffix:?}): {e}",
                        )
                    })?;
                    Ok(TokioRuntimeMode::MultiThreaded { num_workers })
                }
                None => Err(format!("invalid runtime config: {s:?}")),
            },
        }
    }
 }
 static ONE_RUNTIME: Lazy<Option<tokio::runtime::Runtime>> = Lazy::new(|| {
    let thread_name = "pageserver-tokio";
    let Some(mode) = env::var("NEON_PAGESERVER_USE_ONE_RUNTIME") else {
        // If the env var is not set, leave this static as None.
        set_tokio_runtime_setup(
            "multiple-runtimes",
            NUM_MULTIPLE_RUNTIMES
                .checked_mul(*TOKIO_WORKER_THREADS)
                .unwrap(),
        );
        return None;
    };
    Some(match mode {
        TokioRuntimeMode::SingleThreaded => {
            set_tokio_runtime_setup("one-runtime-single-threaded", NonZeroUsize::new(1).unwrap());
            tokio::runtime::Builder::new_current_thread()
                .thread_name(thread_name)
                .enable_all()
                .build()
                .expect("failed to create one single runtime")
        }
        TokioRuntimeMode::MultiThreaded { num_workers } => {
            set_tokio_runtime_setup("one-runtime-multi-threaded", num_workers);
            tokio::runtime::Builder::new_multi_thread()
                .thread_name(thread_name)
                .enable_all()
                .worker_threads(num_workers.get())
                .build()
                .expect("failed to create one multi-threaded runtime")
        }
    })
 });
 /// Declare a lazy static variable named `$varname` that will resolve
 /// to a tokio runtime handle. If the env var `NEON_PAGESERVER_USE_ONE_RUNTIME`
 /// is set, this will resolve to `ONE_RUNTIME`. Otherwise, the macro invocation
 /// declares a separate runtime and the lazy static variable `$varname`
 /// will resolve to that separate runtime.
 ///
 /// The result is is that `$varname.spawn()` will use `ONE_RUNTIME` if
 /// `NEON_PAGESERVER_USE_ONE_RUNTIME` is set, and will use the separate runtime
 /// otherwise.
 macro_rules! pageserver_runtime {
    ($varname:ident, $name:literal) => {
        pub static $varname: Lazy<&'static tokio::runtime::Runtime> = Lazy::new(|| {
            if let Some(runtime) = &*ONE_RUNTIME {
                return runtime;
            }
            static RUNTIME: Lazy<tokio::runtime::Runtime> = Lazy::new(|| {
                tokio::runtime::Builder::new_multi_thread()
                    .thread_name($name)
                    .worker_threads(TOKIO_WORKER_THREADS.get())
                    .enable_all()
                    .build()
                    .expect(std::concat!("Failed to create runtime ", $name))
            });
            &*RUNTIME
        });
    };
 }
 pageserver_runtime!(COMPUTE_REQUEST_RUNTIME, "compute request worker");
 pageserver_runtime!(MGMT_REQUEST_RUNTIME, "mgmt request worker");
 pageserver_runtime!(WALRECEIVER_RUNTIME, "walreceiver worker");
 pageserver_runtime!(BACKGROUND_RUNTIME, "background op worker");
 // Bump this number when adding a new pageserver_runtime!
 // SAFETY: it's obviously correct
 const NUM_MULTIPLE_RUNTIMES: NonZeroUsize = unsafe { NonZeroUsize::new_unchecked(4) };
 #[derive(Debug, Clone, Copy)]
 pub struct PageserverTaskId(u64);
@@ -51,7 +51,7 @@ use tokio_util::sync::CancellationToken;
 use tracing::{info_span, instrument, warn, Instrument};
 use utils::{
    backoff, completion::Barrier, crashsafe::path_with_suffix_extension, failpoint_support, fs_ext,
-    id::TimelineId,
+    id::TimelineId, serde_system_time,
 };
 use super::{
@@ -591,7 +591,7 @@ impl<'a> TenantDownloader<'a> {
        let mut progress = SecondaryProgress {
            layers_total: heatmap_stats.layers,
            bytes_total: heatmap_stats.bytes,
-            heatmap_mtime: Some(heatmap_mtime),
+            heatmap_mtime: Some(serde_system_time::SystemTime(heatmap_mtime)),
            layers_downloaded: 0,
            bytes_downloaded: 0,
        };
@@ -19,6 +19,7 @@ use pageserver_api::models::InMemoryLayerInfo;
 use pageserver_api::shard::TenantShardId;
 use std::collections::{BinaryHeap, HashMap, HashSet};
 use std::sync::{Arc, OnceLock};
 use std::time::Instant;
 use tracing::*;
 use utils::{bin_ser::BeSer, id::TimelineId, lsn::Lsn, vec_map::VecMap};
 // avoid binding to Write (conflicts with std::io::Write)
@@ -53,6 +54,8 @@ pub struct InMemoryLayer {
    /// Writes are only allowed when this is `None`.
    end_lsn: OnceLock<Lsn>,
    opened_at: Instant,
    /// The above fields never change, except for `end_lsn`, which is only set once.
    /// All other changing parts are in `inner`, and protected by a mutex.
    inner: RwLock<InMemoryLayerInner>,
@@ -460,6 +463,7 @@ impl InMemoryLayer {
            tenant_shard_id,
            start_lsn,
            end_lsn: OnceLock::new(),
            opened_at: Instant::now(),
            inner: RwLock::new(InMemoryLayerInner {
                index: HashMap::new(),
                file,
@@ -520,6 +524,10 @@ impl InMemoryLayer {
        Ok(())
    }
    pub(crate) fn get_opened_at(&self) -> Instant {
        self.opened_at
    }
    pub(crate) async fn tick(&self) -> Option<u64> {
        let mut inner = self.inner.write().await;
        let size = inner.file.len();
@@ -18,7 +18,7 @@ use utils::{backoff, completion};
 static CONCURRENT_BACKGROUND_TASKS: once_cell::sync::Lazy<tokio::sync::Semaphore> =
    once_cell::sync::Lazy::new(|| {
-        let total_threads = *task_mgr::BACKGROUND_RUNTIME_WORKER_THREADS;
+        let total_threads = task_mgr::TOKIO_WORKER_THREADS.get();
        let permits = usize::max(
            1,
            // while a lot of the work is done on spawn_blocking, we still do
@@ -72,6 +72,7 @@ pub(crate) async fn concurrent_background_tasks_rate_limit_permit(
        loop_kind == BackgroundLoopKind::InitialLogicalSizeCalculation
    );
    // TODO: assert that we run on BACKGROUND_RUNTIME; requires tokio_unstable Handle::id();
    match CONCURRENT_BACKGROUND_TASKS.acquire().await {
        Ok(permit) => permit,
        Err(_closed) => unreachable!("we never close the semaphore"),
@@ -1257,7 +1257,7 @@ impl Timeline {
            checkpoint_distance,
            self.get_last_record_lsn(),
            self.last_freeze_at.load(),
-            *self.last_freeze_ts.read().unwrap(),
+            open_layer.get_opened_at(),
        ) {
            match open_layer.info() {
                InMemoryLayerInfo::Frozen { lsn_start, lsn_end } => {
@@ -1622,7 +1622,7 @@ impl Timeline {
        checkpoint_distance: u64,
        projected_lsn: Lsn,
        last_freeze_at: Lsn,
-        last_freeze_ts: Instant,
+        opened_at: Instant,
    ) -> bool {
        let distance = projected_lsn.widening_sub(last_freeze_at);
@@ -1648,13 +1648,13 @@ impl Timeline {
            );
            true
-        } else if distance > 0 && last_freeze_ts.elapsed() >= self.get_checkpoint_timeout() {
+        } else if distance > 0 && opened_at.elapsed() >= self.get_checkpoint_timeout() {
            info!(
-                "Will roll layer at {} with layer size {} due to time since last flush ({:?})",
+                    "Will roll layer at {} with layer size {} due to time since first write to the layer ({:?})",
-                projected_lsn,
+                    projected_lsn,
-                layer_size,
+                    layer_size,
-                last_freeze_ts.elapsed()
+                    opened_at.elapsed()
-            );
+                );
            true
        } else {
@@ -4703,23 +4703,16 @@ struct TimelineWriterState {
    max_lsn: Option<Lsn>,
    // Cached details of the last freeze. Avoids going trough the atomic/lock on every put.
    cached_last_freeze_at: Lsn,
    cached_last_freeze_ts: Instant,
 }
 impl TimelineWriterState {
-    fn new(
+    fn new(open_layer: Arc<InMemoryLayer>, current_size: u64, last_freeze_at: Lsn) -> Self {
        open_layer: Arc<InMemoryLayer>,
        current_size: u64,
        last_freeze_at: Lsn,
        last_freeze_ts: Instant,
    ) -> Self {
        Self {
            open_layer,
            current_size,
            prev_lsn: None,
            max_lsn: None,
            cached_last_freeze_at: last_freeze_at,
            cached_last_freeze_ts: last_freeze_ts,
        }
    }
 }
@@ -4818,12 +4811,10 @@ impl<'a> TimelineWriter<'a> {
        let initial_size = layer.size().await?;
        let last_freeze_at = self.last_freeze_at.load();
        let last_freeze_ts = *self.last_freeze_ts.read().unwrap();
        self.write_guard.replace(TimelineWriterState::new(
            layer,
            initial_size,
            last_freeze_at,
            last_freeze_ts,
        ));
        Ok(())
@@ -4870,7 +4861,7 @@ impl<'a> TimelineWriter<'a> {
            self.get_checkpoint_distance(),
            lsn,
            state.cached_last_freeze_at,
-            state.cached_last_freeze_ts,
+            state.open_layer.get_opened_at(),
        ) {
            OpenLayerAction::Roll
        } else {
@@ -12,7 +12,6 @@ use super::layer_manager::LayerManager;
 use super::{CompactFlags, DurationRecorder, RecordedDuration, Timeline};
 use anyhow::{anyhow, Context};
 use async_trait::async_trait;
 use enumset::EnumSet;
 use fail::fail_point;
 use itertools::Itertools;
@@ -1122,7 +1121,6 @@ impl CompactionLayer<Key> for ResidentDeltaLayer {
    }
 }
 #[async_trait]
 impl CompactionDeltaLayer<TimelineAdaptor> for ResidentDeltaLayer {
    type DeltaEntry<'a> = DeltaEntry<'a>;
@@ -41,7 +41,7 @@ pub(crate) fn regenerate(tenants_path: &Path) -> anyhow::Result<PageserverUtiliz
        //
        // note that u64::MAX will be output as i64::MAX as u64, but that should not matter
        utilization_score: u64::MAX,
-        captured_at,
+        captured_at: utils::serde_system_time::SystemTime(captured_at),
    };
    // TODO: make utilization_score into a metric
@@ -12,6 +12,7 @@ testing = []
 anyhow.workspace = true
 async-compression.workspace = true
 async-trait.workspace = true
 atomic-take.workspace = true
 aws-config.workspace = true
 aws-sdk-iam.workspace = true
 aws-sigv4.workspace = true
@@ -36,6 +37,9 @@ http.workspace = true
 humantime.workspace = true
 hyper-tungstenite.workspace = true
 hyper.workspace = true
 hyper1 = { package = "hyper", version = "1.2", features = ["server"] }
 hyper-util = { version = "0.1", features = ["server", "http1", "http2", "tokio"] }
 http-body-util = { version = "0.1" }
 ipnet.workspace = true
 itertools.workspace = true
 lasso = { workspace = true, features = ["multi-threaded"] }
@@ -5,19 +5,13 @@ use std::{
    io,
    net::SocketAddr,
    pin::{pin, Pin},
    sync::Mutex,
    task::{ready, Context, Poll},
 };
 use bytes::{Buf, BytesMut};
-use hyper::server::accept::Accept;
+use hyper::server::conn::AddrIncoming;
 use hyper::server::conn::{AddrIncoming, AddrStream};
 use metrics::IntCounterPairGuard;
 use pin_project_lite::pin_project;
 use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, ReadBuf};
 use uuid::Uuid;
 use crate::metrics::NUM_CLIENT_CONNECTION_GAUGE;
 pub struct ProxyProtocolAccept {
    pub incoming: AddrIncoming,
@@ -331,103 +325,6 @@ impl<T: AsyncRead> AsyncRead for WithClientIp<T> {
    }
 }
 impl Accept for ProxyProtocolAccept {
    type Conn = WithConnectionGuard<WithClientIp<AddrStream>>;
    type Error = io::Error;
    fn poll_accept(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Self::Conn, Self::Error>>> {
        let conn = ready!(Pin::new(&mut self.incoming).poll_accept(cx)?);
        let conn_id = uuid::Uuid::new_v4();
        let span = tracing::info_span!("http_conn", ?conn_id);
        {
            let _enter = span.enter();
            tracing::info!("accepted new TCP connection");
        }
        let Some(conn) = conn else {
            return Poll::Ready(None);
        };
        Poll::Ready(Some(Ok(WithConnectionGuard {
            inner: WithClientIp::new(conn),
            connection_id: Uuid::new_v4(),
            gauge: Mutex::new(Some(
                NUM_CLIENT_CONNECTION_GAUGE
                    .with_label_values(&[self.protocol])
                    .guard(),
            )),
            span,
        })))
    }
 }
 pin_project! {
    pub struct WithConnectionGuard<T> {
        #[pin]
        pub inner: T,
        pub connection_id: Uuid,
        pub gauge: Mutex<Option<IntCounterPairGuard>>,
        pub span: tracing::Span,
    }
    impl<S> PinnedDrop for WithConnectionGuard<S> {
        fn drop(this: Pin<&mut Self>) {
            let _enter = this.span.enter();
            tracing::info!("HTTP connection closed")
        }
    }
 }
 impl<T: AsyncWrite> AsyncWrite for WithConnectionGuard<T> {
    #[inline]
    fn poll_write(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<Result<usize, io::Error>> {
        self.project().inner.poll_write(cx, buf)
    }
    #[inline]
    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
        self.project().inner.poll_flush(cx)
    }
    #[inline]
    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
        self.project().inner.poll_shutdown(cx)
    }
    #[inline]
    fn poll_write_vectored(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        bufs: &[io::IoSlice<'_>],
    ) -> Poll<Result<usize, io::Error>> {
        self.project().inner.poll_write_vectored(cx, bufs)
    }
    #[inline]
    fn is_write_vectored(&self) -> bool {
        self.inner.is_write_vectored()
    }
 }
 impl<T: AsyncRead> AsyncRead for WithConnectionGuard<T> {
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<io::Result<()>> {
        self.project().inner.poll_read(cx, buf)
    }
 }
 #[cfg(test)]
 mod tests {
    use std::pin::pin;
@@ -4,42 +4,48 @@
 mod backend;
 mod conn_pool;
 mod http_util;
 mod json;
 mod sql_over_http;
 pub mod tls_listener;
 mod websocket;
 use atomic_take::AtomicTake;
 use bytes::Bytes;
 pub use conn_pool::GlobalConnPoolOptions;
-use anyhow::bail;
+use anyhow::Context;
-use hyper::StatusCode;
+use futures::future::{select, Either};
-use metrics::IntCounterPairGuard;
+use futures::TryFutureExt;
 use http::{Method, Response, StatusCode};
 use http_body_util::Full;
 use hyper1::body::Incoming;
 use hyper_util::rt::TokioExecutor;
 use hyper_util::server::conn::auto::Builder;
 use rand::rngs::StdRng;
 use rand::SeedableRng;
 pub use reqwest_middleware::{ClientWithMiddleware, Error};
 pub use reqwest_retry::{policies::ExponentialBackoff, RetryTransientMiddleware};
 use tokio::time::timeout;
 use tokio_rustls::TlsAcceptor;
 use tokio_util::task::TaskTracker;
 use tracing::instrument::Instrumented;
 use crate::cancellation::CancellationHandlerMain;
 use crate::config::ProxyConfig;
 use crate::context::RequestMonitoring;
-use crate::protocol2::{ProxyProtocolAccept, WithClientIp, WithConnectionGuard};
+use crate::metrics::{NUM_CLIENT_CONNECTION_GAUGE, TLS_HANDSHAKE_FAILURES};
 use crate::protocol2::WithClientIp;
 use crate::proxy::run_until_cancelled;
 use crate::rate_limiter::EndpointRateLimiter;
 use crate::serverless::backend::PoolingBackend;
-use hyper::{
+use crate::serverless::http_util::{api_error_into_response, json_response};
    server::conn::{AddrIncoming, AddrStream},
    Body, Method, Request, Response,
 };
-use std::net::IpAddr;
+use std::net::{IpAddr, SocketAddr};
 use std::pin::pin;
 use std::sync::Arc;
-use std::task::Poll;
+use tokio::net::{TcpListener, TcpStream};
-use tls_listener::TlsListener;
+use tokio_util::sync::CancellationToken;
 use tokio::net::TcpListener;
 use tokio_util::sync::{CancellationToken, DropGuard};
 use tracing::{error, info, warn, Instrument};
-use utils::http::{error::ApiError, json::json_response};
+use utils::http::error::ApiError;
 pub const SERVERLESS_DRIVER_SNI: &str = "api";
@@ -91,161 +97,174 @@ pub async fn task_main(
    tls_server_config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
    let tls_acceptor: tokio_rustls::TlsAcceptor = Arc::new(tls_server_config).into();
-    let mut addr_incoming = AddrIncoming::from_listener(ws_listener)?;
+    let connections = tokio_util::task::task_tracker::TaskTracker::new();
-    let _ = addr_incoming.set_nodelay(true);
+    connections.close(); // allows `connections.wait to complete`
    let addr_incoming = ProxyProtocolAccept {
        incoming: addr_incoming,
        protocol: "http",
    };
-    let ws_connections = tokio_util::task::task_tracker::TaskTracker::new();
+    let server = Builder::new(hyper_util::rt::TokioExecutor::new());
    ws_connections.close(); // allows `ws_connections.wait to complete`
-    let tls_listener = TlsListener::new(tls_acceptor, addr_incoming, config.handshake_timeout);
+    while let Some(res) = run_until_cancelled(ws_listener.accept(), &cancellation_token).await {
        let (conn, peer_addr) = res.context("could not accept TCP stream")?;
        if let Err(e) = conn.set_nodelay(true) {
            tracing::error!("could not set nodelay: {e}");
            continue;
        }
        let conn_id = uuid::Uuid::new_v4();
        let http_conn_span = tracing::info_span!("http_conn", ?conn_id);
-    let make_svc = hyper::service::make_service_fn(
+        connections.spawn(
-        |stream: &tokio_rustls::server::TlsStream<
+            connection_handler(
-            WithConnectionGuard<WithClientIp<AddrStream>>,
+                config,
-        >| {
+                backend.clone(),
-            let (conn, _) = stream.get_ref();
+                connections.clone(),
                cancellation_handler.clone(),
                endpoint_rate_limiter.clone(),
                cancellation_token.clone(),
                server.clone(),
                tls_acceptor.clone(),
                conn,
                peer_addr,
            )
            .instrument(http_conn_span),
        );
    }
-            // this is jank. should dissapear with hyper 1.0 migration.
+    connections.wait().await;
            let gauge = conn
                .gauge
                .lock()
                .expect("lock should not be poisoned")
                .take()
                .expect("gauge should be set on connection start");
            // Cancel all current inflight HTTP requests if the HTTP connection is closed.
            let http_cancellation_token = CancellationToken::new();
            let cancel_connection = http_cancellation_token.clone().drop_guard();
            let span = conn.span.clone();
            let client_addr = conn.inner.client_addr();
            let remote_addr = conn.inner.inner.remote_addr();
            let backend = backend.clone();
            let ws_connections = ws_connections.clone();
            let endpoint_rate_limiter = endpoint_rate_limiter.clone();
            let cancellation_handler = cancellation_handler.clone();
            async move {
                let peer_addr = match client_addr {
                    Some(addr) => addr,
                    None if config.require_client_ip => bail!("missing required client ip"),
                    None => remote_addr,
                };
                Ok(MetricService::new(
                    hyper::service::service_fn(move |req: Request<Body>| {
                        let backend = backend.clone();
                        let ws_connections2 = ws_connections.clone();
                        let endpoint_rate_limiter = endpoint_rate_limiter.clone();
                        let cancellation_handler = cancellation_handler.clone();
                        let http_cancellation_token = http_cancellation_token.child_token();
                        // `request_handler` is not cancel safe. It expects to be cancelled only at specific times.
                        // By spawning the future, we ensure it never gets cancelled until it decides to.
                        ws_connections.spawn(
                            async move {
                                // Cancel the current inflight HTTP request if the requets stream is closed.
                                // This is slightly different to `_cancel_connection` in that
                                // h2 can cancel individual requests with a `RST_STREAM`.
                                let _cancel_session = http_cancellation_token.clone().drop_guard();
                                let res = request_handler(
                                    req,
                                    config,
                                    backend,
                                    ws_connections2,
                                    cancellation_handler,
                                    peer_addr.ip(),
                                    endpoint_rate_limiter,
                                    http_cancellation_token,
                                )
                                .await
                                .map_or_else(|e| e.into_response(), |r| r);
                                _cancel_session.disarm();
                                res
                            }
                            .in_current_span(),
                        )
                    }),
                    gauge,
                    cancel_connection,
                    span,
                ))
            }
        },
    );
    hyper::Server::builder(tls_listener)
        .serve(make_svc)
        .with_graceful_shutdown(cancellation_token.cancelled())
        .await?;
    // await websocket connections
    ws_connections.wait().await;
    Ok(())
 }
-struct MetricService<S> {
+/// Handles the TCP lifecycle.
-    inner: S,
+///
-    _gauge: IntCounterPairGuard,
+/// 1. Parses PROXY protocol V2
-    _cancel: DropGuard,
+/// 2. Handles TLS handshake
-    span: tracing::Span,
+/// 3. Handles HTTP connection
-}
+///     1. With graceful shutdowns
 ///     2. With graceful request cancellation with connection failure
 ///     3. With websocket upgrade support.
 #[allow(clippy::too_many_arguments)]
 async fn connection_handler(
    config: &'static ProxyConfig,
    backend: Arc<PoolingBackend>,
    connections: TaskTracker,
    cancellation_handler: Arc<CancellationHandlerMain>,
    endpoint_rate_limiter: Arc<EndpointRateLimiter>,
    cancellation_token: CancellationToken,
    server: Builder<TokioExecutor>,
    tls_acceptor: TlsAcceptor,
    conn: TcpStream,
    peer_addr: SocketAddr,
 ) {
    let session_id = uuid::Uuid::new_v4();
-impl<S> MetricService<S> {
+    let _gauge = NUM_CLIENT_CONNECTION_GAUGE
-    fn new(
+        .with_label_values(&["http"])
-        inner: S,
+        .guard();
-        _gauge: IntCounterPairGuard,
+
-        _cancel: DropGuard,
+    // handle PROXY protocol
-        span: tracing::Span,
+    let mut conn = WithClientIp::new(conn);
-    ) -> MetricService<S> {
+    let peer = match conn.wait_for_addr().await {
-        MetricService {
+        Ok(peer) => peer,
-            inner,
+        Err(e) => {
-            _gauge,
+            tracing::error!(?session_id, %peer_addr, "failed to accept TCP connection: invalid PROXY protocol V2 header: {e:#}");
-            _cancel,
+            return;
            span,
        }
-    }
+    };
 }
-impl<S, ReqBody> hyper::service::Service<Request<ReqBody>> for MetricService<S>
+    let peer_addr = peer.unwrap_or(peer_addr).ip();
-where
+    info!(?session_id, %peer_addr, "accepted new TCP connection");
    S: hyper::service::Service<Request<ReqBody>>,
 {
    type Response = S::Response;
    type Error = S::Error;
    type Future = Instrumented<S::Future>;
-    fn poll_ready(&mut self, cx: &mut std::task::Context<'_>) -> Poll<Result<(), Self::Error>> {
+    // try upgrade to TLS, but with a timeout.
-        self.inner.poll_ready(cx)
+    let conn = match timeout(config.handshake_timeout, tls_acceptor.accept(conn)).await {
-    }
+        Ok(Ok(conn)) => {
            info!(?session_id, %peer_addr, "accepted new TLS connection");
            conn
        }
        // The handshake failed
        Ok(Err(e)) => {
            TLS_HANDSHAKE_FAILURES.inc();
            warn!(?session_id, %peer_addr, "failed to accept TLS connection: {e:?}");
            return;
        }
        // The handshake timed out
        Err(e) => {
            TLS_HANDSHAKE_FAILURES.inc();
            warn!(?session_id, %peer_addr, "failed to accept TLS connection: {e:?}");
            return;
        }
    };
-    fn call(&mut self, req: Request<ReqBody>) -> Self::Future {
+    let session_id = AtomicTake::new(session_id);
-        self.span
+
-            .in_scope(|| self.inner.call(req))
+    // Cancel all current inflight HTTP requests if the HTTP connection is closed.
-            .instrument(self.span.clone())
+    let http_cancellation_token = CancellationToken::new();
    let _cancel_connection = http_cancellation_token.clone().drop_guard();
    let conn = server.serve_connection_with_upgrades(
        hyper_util::rt::TokioIo::new(conn),
        hyper1::service::service_fn(move |req: hyper1::Request<Incoming>| {
            // First HTTP request shares the same session ID
            let session_id = session_id.take().unwrap_or_else(uuid::Uuid::new_v4);
            // Cancel the current inflight HTTP request if the requets stream is closed.
            // This is slightly different to `_cancel_connection` in that
            // h2 can cancel individual requests with a `RST_STREAM`.
            let http_request_token = http_cancellation_token.child_token();
            let cancel_request = http_request_token.clone().drop_guard();
            // `request_handler` is not cancel safe. It expects to be cancelled only at specific times.
            // By spawning the future, we ensure it never gets cancelled until it decides to.
            let handler = connections.spawn(
                request_handler(
                    req,
                    config,
                    backend.clone(),
                    connections.clone(),
                    cancellation_handler.clone(),
                    session_id,
                    peer_addr,
                    endpoint_rate_limiter.clone(),
                    http_request_token,
                )
                .in_current_span()
                .map_ok_or_else(api_error_into_response, |r| r),
            );
            async move {
                let res = handler.await;
                cancel_request.disarm();
                res
            }
        }),
    );
    // On cancellation, trigger the HTTP connection handler to shut down.
    let res = match select(pin!(cancellation_token.cancelled()), pin!(conn)).await {
        Either::Left((_cancelled, mut conn)) => {
            conn.as_mut().graceful_shutdown();
            conn.await
        }
        Either::Right((res, _)) => res,
    };
    match res {
        Ok(()) => tracing::info!(%peer_addr, "HTTP connection closed"),
        Err(e) => tracing::warn!(%peer_addr, "HTTP connection error {e}"),
    }
 }
 #[allow(clippy::too_many_arguments)]
 async fn request_handler(
-    mut request: Request<Body>,
+    mut request: hyper1::Request<Incoming>,
    config: &'static ProxyConfig,
    backend: Arc<PoolingBackend>,
    ws_connections: TaskTracker,
    cancellation_handler: Arc<CancellationHandlerMain>,
    session_id: uuid::Uuid,
    peer_addr: IpAddr,
    endpoint_rate_limiter: Arc<EndpointRateLimiter>,
    // used to cancel in-flight HTTP requests. not used to cancel websockets
    http_cancellation_token: CancellationToken,
-) -> Result<Response<Body>, ApiError> {
+) -> Result<Response<Full<Bytes>>, ApiError> {
    let session_id = uuid::Uuid::new_v4();
    let host = request
        .headers()
        .get("host")
@@ -282,14 +301,14 @@ async fn request_handler(
        // Return the response so the spawned future can continue.
        Ok(response)
-    } else if request.uri().path() == "/sql" && request.method() == Method::POST {
+    } else if request.uri().path() == "/sql" && *request.method() == Method::POST {
        let ctx = RequestMonitoring::new(session_id, peer_addr, "http", &config.region);
        let span = ctx.span.clone();
        sql_over_http::handle(config, ctx, request, backend, http_cancellation_token)
            .instrument(span)
            .await
-    } else if request.uri().path() == "/sql" && request.method() == Method::OPTIONS {
+    } else if request.uri().path() == "/sql" && *request.method() == Method::OPTIONS {
        Response::builder()
            .header("Allow", "OPTIONS, POST")
            .header("Access-Control-Allow-Origin", "*")
@@ -299,7 +318,7 @@ async fn request_handler(
            )
            .header("Access-Control-Max-Age", "86400" /* 24 hours */)
            .status(StatusCode::OK) // 204 is also valid, but see: https://developer.mozilla.org/en-US/docs/Web/HTTP/Methods/OPTIONS#status_code
-            .body(Body::empty())
+            .body(Full::new(Bytes::new()))
            .map_err(|e| ApiError::InternalServerError(e.into()))
    } else {
        json_response(StatusCode::BAD_REQUEST, "query is not supported")
@@ -0,0 +1,92 @@
 //! Things stolen from `libs/utils/src/http` to add hyper 1.0 compatibility
 //! Will merge back in at some point in the future.
 use bytes::Bytes;
 use anyhow::Context;
 use http::{Response, StatusCode};
 use http_body_util::Full;
 use serde::Serialize;
 use utils::http::error::ApiError;
 /// Like [`ApiError::into_response`]
 pub fn api_error_into_response(this: ApiError) -> Response<Full<Bytes>> {
    match this {
        ApiError::BadRequest(err) => HttpErrorBody::response_from_msg_and_status(
            format!("{err:#?}"), // use debug printing so that we give the cause
            StatusCode::BAD_REQUEST,
        ),
        ApiError::Forbidden(_) => {
            HttpErrorBody::response_from_msg_and_status(this.to_string(), StatusCode::FORBIDDEN)
        }
        ApiError::Unauthorized(_) => {
            HttpErrorBody::response_from_msg_and_status(this.to_string(), StatusCode::UNAUTHORIZED)
        }
        ApiError::NotFound(_) => {
            HttpErrorBody::response_from_msg_and_status(this.to_string(), StatusCode::NOT_FOUND)
        }
        ApiError::Conflict(_) => {
            HttpErrorBody::response_from_msg_and_status(this.to_string(), StatusCode::CONFLICT)
        }
        ApiError::PreconditionFailed(_) => HttpErrorBody::response_from_msg_and_status(
            this.to_string(),
            StatusCode::PRECONDITION_FAILED,
        ),
        ApiError::ShuttingDown => HttpErrorBody::response_from_msg_and_status(
            "Shutting down".to_string(),
            StatusCode::SERVICE_UNAVAILABLE,
        ),
        ApiError::ResourceUnavailable(err) => HttpErrorBody::response_from_msg_and_status(
            err.to_string(),
            StatusCode::SERVICE_UNAVAILABLE,
        ),
        ApiError::Timeout(err) => HttpErrorBody::response_from_msg_and_status(
            err.to_string(),
            StatusCode::REQUEST_TIMEOUT,
        ),
        ApiError::InternalServerError(err) => HttpErrorBody::response_from_msg_and_status(
            err.to_string(),
            StatusCode::INTERNAL_SERVER_ERROR,
        ),
    }
 }
 /// Same as [`utils::http::error::HttpErrorBody`]
 #[derive(Serialize)]
 struct HttpErrorBody {
    pub msg: String,
 }
 impl HttpErrorBody {
    /// Same as [`utils::http::error::HttpErrorBody::response_from_msg_and_status`]
    fn response_from_msg_and_status(msg: String, status: StatusCode) -> Response<Full<Bytes>> {
        HttpErrorBody { msg }.to_response(status)
    }
    /// Same as [`utils::http::error::HttpErrorBody::to_response`]
    fn to_response(&self, status: StatusCode) -> Response<Full<Bytes>> {
        Response::builder()
            .status(status)
            .header(http::header::CONTENT_TYPE, "application/json")
            // we do not have nested maps with non string keys so serialization shouldn't fail
            .body(Full::new(Bytes::from(serde_json::to_string(self).unwrap())))
            .unwrap()
    }
 }
 /// Same as [`utils::http::json::json_response`]
 pub fn json_response<T: Serialize>(
    status: StatusCode,
    data: T,
 ) -> Result<Response<Full<Bytes>>, ApiError> {
    let json = serde_json::to_string(&data)
        .context("Failed to serialize JSON response")
        .map_err(ApiError::InternalServerError)?;
    let response = Response::builder()
        .status(status)
        .header(http::header::CONTENT_TYPE, "application/json")
        .body(Full::new(Bytes::from(json)))
        .map_err(|e| ApiError::InternalServerError(e.into()))?;
    Ok(response)
 }
@@ -1,18 +1,22 @@
 use std::pin::pin;
 use std::sync::Arc;
 use bytes::Bytes;
 use futures::future::select;
 use futures::future::try_join;
 use futures::future::Either;
 use futures::StreamExt;
 use futures::TryFutureExt;
-use hyper::body::HttpBody;
+use http_body_util::BodyExt;
-use hyper::header;
+use http_body_util::Full;
-use hyper::http::HeaderName;
+use hyper1::body::Body;
-use hyper::http::HeaderValue;
+use hyper1::body::Incoming;
-use hyper::Response;
+use hyper1::header;
-use hyper::StatusCode;
+use hyper1::http::HeaderName;
-use hyper::{Body, HeaderMap, Request};
+use hyper1::http::HeaderValue;
 use hyper1::Response;
 use hyper1::StatusCode;
 use hyper1::{HeaderMap, Request};
 use serde_json::json;
 use serde_json::Value;
 use tokio::time;
@@ -29,7 +33,6 @@ use tracing::error;
 use tracing::info;
 use url::Url;
 use utils::http::error::ApiError;
 use utils::http::json::json_response;
 use crate::auth::backend::ComputeUserInfo;
 use crate::auth::endpoint_sni;
@@ -52,6 +55,7 @@ use crate::RoleName;
 use super::backend::PoolingBackend;
 use super::conn_pool::Client;
 use super::conn_pool::ConnInfo;
 use super::http_util::json_response;
 use super::json::json_to_pg_text;
 use super::json::pg_text_row_to_json;
 use super::json::JsonConversionError;
@@ -218,10 +222,10 @@ fn get_conn_info(
 pub async fn handle(
    config: &'static ProxyConfig,
    mut ctx: RequestMonitoring,
-    request: Request<Body>,
+    request: Request<Incoming>,
    backend: Arc<PoolingBackend>,
    cancel: CancellationToken,
-) -> Result<Response<Body>, ApiError> {
+) -> Result<Response<Full<Bytes>>, ApiError> {
    let result = handle_inner(cancel, config, &mut ctx, request, backend).await;
    let mut response = match result {
@@ -332,10 +336,9 @@ pub async fn handle(
        }
    };
-    response.headers_mut().insert(
+    response
-        "Access-Control-Allow-Origin",
+        .headers_mut()
-        hyper::http::HeaderValue::from_static("*"),
+        .insert("Access-Control-Allow-Origin", HeaderValue::from_static("*"));
    );
    Ok(response)
 }
@@ -396,7 +399,7 @@ impl UserFacingError for SqlOverHttpError {
 #[derive(Debug, thiserror::Error)]
 pub enum ReadPayloadError {
    #[error("could not read the HTTP request body: {0}")]
-    Read(#[from] hyper::Error),
+    Read(#[from] hyper1::Error),
    #[error("could not parse the HTTP request body: {0}")]
    Parse(#[from] serde_json::Error),
 }
@@ -437,7 +440,7 @@ struct HttpHeaders {
 }
 impl HttpHeaders {
-    fn try_parse(headers: &hyper::http::HeaderMap) -> Result<Self, SqlOverHttpError> {
+    fn try_parse(headers: &hyper1::http::HeaderMap) -> Result<Self, SqlOverHttpError> {
        // Determine the output options. Default behaviour is 'false'. Anything that is not
        // strictly 'true' assumed to be false.
        let raw_output = headers.get(&RAW_TEXT_OUTPUT) == Some(&HEADER_VALUE_TRUE);
@@ -488,9 +491,9 @@ async fn handle_inner(
    cancel: CancellationToken,
    config: &'static ProxyConfig,
    ctx: &mut RequestMonitoring,
-    request: Request<Body>,
+    request: Request<Incoming>,
    backend: Arc<PoolingBackend>,
-) -> Result<Response<Body>, SqlOverHttpError> {
+) -> Result<Response<Full<Bytes>>, SqlOverHttpError> {
    let _request_gauge = NUM_CONNECTION_REQUESTS_GAUGE
        .with_label_values(&[ctx.protocol])
        .guard();
@@ -528,7 +531,7 @@ async fn handle_inner(
    }
    let fetch_and_process_request = async {
-        let body = hyper::body::to_bytes(request.into_body()).await?;
+        let body = request.into_body().collect().await?.to_bytes();
        info!(length = body.len(), "request payload read");
        let payload: Payload = serde_json::from_slice(&body)?;
        Ok::<Payload, ReadPayloadError>(payload) // Adjust error type accordingly
@@ -596,7 +599,7 @@ async fn handle_inner(
    let body = serde_json::to_string(&result).expect("json serialization should not fail");
    let len = body.len();
    let response = response
-        .body(Body::from(body))
+        .body(Full::new(Bytes::from(body)))
        // only fails if invalid status code or invalid header/values are given.
        // these are not user configurable so it cannot fail dynamically
        .expect("building response payload should not fail");
@@ -639,6 +642,7 @@ impl QueryData {
            }
            // The query was cancelled.
            Either::Right((_cancelled, query)) => {
                tracing::info!("cancelling query");
                if let Err(err) = cancel_token.cancel_query(NoTls).await {
                    tracing::error!(?err, "could not cancel query");
                }
@@ -1,123 +0,0 @@
 use std::{
    convert::Infallible,
    pin::Pin,
    task::{Context, Poll},
    time::Duration,
 };
 use hyper::server::{accept::Accept, conn::AddrStream};
 use pin_project_lite::pin_project;
 use tokio::{
    io::{AsyncRead, AsyncWrite},
    task::JoinSet,
    time::timeout,
 };
 use tokio_rustls::{server::TlsStream, TlsAcceptor};
 use tracing::{info, warn, Instrument};
 use crate::{
    metrics::TLS_HANDSHAKE_FAILURES,
    protocol2::{WithClientIp, WithConnectionGuard},
 };
 pin_project! {
    /// Wraps a `Stream` of connections (such as a TCP listener) so that each connection is itself
    /// encrypted using TLS.
    pub(crate) struct TlsListener<A: Accept> {
        #[pin]
        listener: A,
        tls: TlsAcceptor,
        waiting: JoinSet<Option<TlsStream<A::Conn>>>,
        timeout: Duration,
    }
 }
 impl<A: Accept> TlsListener<A> {
    /// Create a `TlsListener` with default options.
    pub(crate) fn new(tls: TlsAcceptor, listener: A, timeout: Duration) -> Self {
        TlsListener {
            listener,
            tls,
            waiting: JoinSet::new(),
            timeout,
        }
    }
 }
 impl<A> Accept for TlsListener<A>
 where
    A: Accept<Conn = WithConnectionGuard<WithClientIp<AddrStream>>>,
    A::Error: std::error::Error,
    A::Conn: AsyncRead + AsyncWrite + Unpin + Send + 'static,
 {
    type Conn = TlsStream<A::Conn>;
    type Error = Infallible;
    fn poll_accept(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Self::Conn, Self::Error>>> {
        let mut this = self.project();
        loop {
            match this.listener.as_mut().poll_accept(cx) {
                Poll::Pending => break,
                Poll::Ready(Some(Ok(mut conn))) => {
                    let t = *this.timeout;
                    let tls = this.tls.clone();
                    let span = conn.span.clone();
                    this.waiting.spawn(async move {
                        let peer_addr = match conn.inner.wait_for_addr().await {
                            Ok(Some(addr)) => addr,
                            Err(e) => {
                                tracing::error!("failed to accept TCP connection: invalid PROXY protocol V2 header: {e:#}");
                                return None;
                            }
                            Ok(None) => conn.inner.inner.remote_addr()
                        };
                        let accept = tls.accept(conn);
                        match timeout(t, accept).await {
                            Ok(Ok(conn)) => {
                                info!(%peer_addr, "accepted new TLS connection");
                                Some(conn)
                            },
                            // The handshake failed, try getting another connection from the queue
                            Ok(Err(e)) => {
                                TLS_HANDSHAKE_FAILURES.inc();
                                warn!(%peer_addr, "failed to accept TLS connection: {e:?}");
                                None
                            }
                            // The handshake timed out, try getting another connection from the queue
                            Err(_) => {
                                TLS_HANDSHAKE_FAILURES.inc();
                                warn!(%peer_addr, "failed to accept TLS connection: timeout");
                                None
                            }
                        }
                    }.instrument(span));
                }
                Poll::Ready(Some(Err(e))) => {
                    tracing::error!("error accepting TCP connection: {e}");
                    continue;
                }
                Poll::Ready(None) => return Poll::Ready(None),
            }
        }
        loop {
            return match this.waiting.poll_join_next(cx) {
                Poll::Ready(Some(Ok(Some(conn)))) => Poll::Ready(Some(Ok(conn))),
                // The handshake failed to complete, try getting another connection from the queue
                Poll::Ready(Some(Ok(None))) => continue,
                // The handshake panicked or was cancelled. ignore and get another connection
                Poll::Ready(Some(Err(e))) => {
                    tracing::warn!("handshake aborted: {e}");
                    continue;
                }
                _ => Poll::Pending,
            };
        }
    }
 }
@@ -15,8 +15,7 @@ FLAKY_TESTS_QUERY = """
        DISTINCT parent_suite, suite, name
    FROM results
    WHERE
-        started_at > CURRENT_DATE - INTERVAL '10' day
+        started_at > CURRENT_DATE - INTERVAL '%s' day
        AND started_at > '2024-03-11 14:50:11.845+00' -- we switched the default PAGESERVER_VIRTUAL_FILE_IO_ENGINE to `tokio-epoll-uring` from `std-fs` on this date, we want to ignore the flaky tests for `std-fs`
        AND (
            (status IN ('failed', 'broken') AND reference = 'refs/heads/main')
            OR flaky
@@ -22,7 +22,7 @@ parser.add_argument("--safekeeper-host", required=True, type=str)
 args = parser.parse_args()
 access_key = os.getenv("CONSOLE_API_TOKEN")
-endpoint: str = "https://console.stage.neon.tech/api"
+endpoint: str = "https://console-stage.neon.build/api"
 trash_dir: Path = args.trash_dir
 dry_run: bool = args.dry_run
@@ -3,7 +3,7 @@
 3. Issue admin token (add/remove .stage from url for staging/prod and setting proper API key):
 ```
 # staging:
-AUTH_TOKEN=$(curl https://console.stage.neon.tech/regions/console/api/v1/admin/issue_token -H "Accept: application/json" -H "Content-Type: application/json" -H "Authorization: Bearer $NEON_STAGING_KEY" -X POST -d '{"ttl_seconds": 43200, "scope": "safekeeperdata"}' 2>/dev/null | jq --raw-output '.jwt')
+AUTH_TOKEN=$(curl https://console-stage.neon.build/regions/console/api/v1/admin/issue_token -H "Accept: application/json" -H "Content-Type: application/json" -H "Authorization: Bearer $NEON_STAGING_KEY" -X POST -d '{"ttl_seconds": 43200, "scope": "safekeeperdata"}' 2>/dev/null | jq --raw-output '.jwt')
 # prod:
 AUTH_TOKEN=$(curl https://console.neon.tech/regions/console/api/v1/admin/issue_token -H "Accept: application/json" -H "Content-Type: application/json" -H "Authorization: Bearer $NEON_PROD_KEY" -X POST -d '{"ttl_seconds": 43200, "scope": "safekeeperdata"}' 2>/dev/null | jq --raw-output '.jwt')
 # check
@@ -8,6 +8,7 @@ use futures::Future;
 use hyper::header::CONTENT_TYPE;
 use hyper::{Body, Request, Response};
 use hyper::{StatusCode, Uri};
 use metrics::{BuildInfo, NeonMetrics};
 use pageserver_api::models::{
    TenantConfigRequest, TenantCreateRequest, TenantLocationConfigRequest, TenantShardSplitRequest,
    TenantTimeTravelRequest, TimelineCreateRequest,
@@ -44,15 +45,19 @@ use control_plane::storage_controller::{AttachHookRequest, InspectRequest};
 use routerify::Middleware;
 /// State available to HTTP request handlers
 #[derive(Clone)]
 pub struct HttpState {
    service: Arc<crate::service::Service>,
    auth: Option<Arc<SwappableJwtAuth>>,
    neon_metrics: NeonMetrics,
    allowlist_routes: Vec<Uri>,
 }
 impl HttpState {
-    pub fn new(service: Arc<crate::service::Service>, auth: Option<Arc<SwappableJwtAuth>>) -> Self {
+    pub fn new(
        service: Arc<crate::service::Service>,
        auth: Option<Arc<SwappableJwtAuth>>,
        build_info: BuildInfo,
    ) -> Self {
        let allowlist_routes = ["/status", "/ready", "/metrics"]
            .iter()
            .map(|v| v.parse().unwrap())
@@ -60,6 +65,7 @@ impl HttpState {
        Self {
            service,
            auth,
            neon_metrics: NeonMetrics::new(build_info),
            allowlist_routes,
        }
    }
@@ -672,10 +678,11 @@ fn epilogue_metrics_middleware<B: hyper::body::HttpBody + Send + Sync + 'static>
    })
 }
-pub async fn measured_metrics_handler(_req: Request<Body>) -> Result<Response<Body>, ApiError> {
+pub async fn measured_metrics_handler(req: Request<Body>) -> Result<Response<Body>, ApiError> {
    pub const TEXT_FORMAT: &str = "text/plain; version=0.0.4";
-    let payload = crate::metrics::METRICS_REGISTRY.encode();
+    let state = get_state(&req);
    let payload = crate::metrics::METRICS_REGISTRY.encode(&state.neon_metrics);
    let response = Response::builder()
        .status(200)
        .header(CONTENT_TYPE, TEXT_FORMAT)
@@ -704,6 +711,7 @@ where
 pub fn make_router(
    service: Arc<Service>,
    auth: Option<Arc<SwappableJwtAuth>>,
    build_info: BuildInfo,
 ) -> RouterBuilder<hyper::Body, ApiError> {
    let mut router = endpoint::make_router()
        .middleware(prologue_metrics_middleware())
@@ -720,7 +728,7 @@ pub fn make_router(
    }
    router
-        .data(Arc::new(HttpState::new(service, auth)))
+        .data(Arc::new(HttpState::new(service, auth, build_info)))
        .get("/metrics", |r| {
            named_request_span(r, measured_metrics_handler, RequestName("metrics"))
        })
@@ -3,6 +3,7 @@ use camino::Utf8PathBuf;
 use clap::Parser;
 use diesel::Connection;
 use metrics::launch_timestamp::LaunchTimestamp;
 use metrics::BuildInfo;
 use std::sync::Arc;
 use storage_controller::http::make_router;
 use storage_controller::metrics::preinitialize_metrics;
@@ -192,6 +193,11 @@ async fn async_main() -> anyhow::Result<()> {
        args.listen
    );
    let build_info = BuildInfo {
        revision: GIT_VERSION,
        build_tag: BUILD_TAG,
    };
    let strict_mode = if args.dev {
        StrictMode::Dev
    } else {
@@ -253,7 +259,7 @@ async fn async_main() -> anyhow::Result<()> {
    let auth = secrets
        .public_key
        .map(|jwt_auth| Arc::new(SwappableJwtAuth::new(jwt_auth)));
-    let router = make_router(service.clone(), auth)
+    let router = make_router(service.clone(), auth, build_info)
        .build()
        .map_err(|err| anyhow!(err))?;
    let router_service = utils::http::RouterService::new(router).unwrap();
@@ -8,10 +8,8 @@
 //! The rest of the code defines label group types and deals with converting outer types to labels.
 //!
 use bytes::Bytes;
-use measured::{
+use measured::{label::LabelValue, metric::histogram, FixedCardinalityLabel, MetricGroup};
-    label::{LabelValue, StaticLabelSet},
+use metrics::NeonMetrics;
    FixedCardinalityLabel, MetricGroup,
 };
 use once_cell::sync::Lazy;
 use std::sync::Mutex;
@@ -26,13 +24,15 @@ pub fn preinitialize_metrics() {
 pub(crate) struct StorageControllerMetrics {
    pub(crate) metrics_group: StorageControllerMetricGroup,
-    encoder: Mutex<measured::text::TextEncoder>,
+    encoder: Mutex<measured::text::BufferedTextEncoder>,
 }
 #[derive(measured::MetricGroup)]
 #[metric(new())]
 pub(crate) struct StorageControllerMetricGroup {
    /// Count of how many times we spawn a reconcile task
    pub(crate) storage_controller_reconcile_spawn: measured::Counter,
    /// Reconciler tasks completed, broken down by success/failure/cancelled
    pub(crate) storage_controller_reconcile_complete:
        measured::CounterVec<ReconcileCompleteLabelGroupSet>,
@@ -43,7 +43,9 @@ pub(crate) struct StorageControllerMetricGroup {
    /// HTTP request status counters for handled requests
    pub(crate) storage_controller_http_request_status:
        measured::CounterVec<HttpRequestStatusLabelGroupSet>,
    /// HTTP request handler latency across all status codes
    #[metric(metadata = histogram::Thresholds::exponential_buckets(0.1, 2.0))]
    pub(crate) storage_controller_http_request_latency:
        measured::HistogramVec<HttpRequestLatencyLabelGroupSet, 5>,
@@ -55,6 +57,7 @@ pub(crate) struct StorageControllerMetricGroup {
    /// Latency of HTTP requests to the pageserver, broken down by pageserver
    /// node id, request name and method. This include both successful and unsuccessful
    /// requests.
    #[metric(metadata = histogram::Thresholds::exponential_buckets(0.1, 2.0))]
    pub(crate) storage_controller_pageserver_request_latency:
        measured::HistogramVec<PageserverRequestLabelGroupSet, 5>,
@@ -66,6 +69,7 @@ pub(crate) struct StorageControllerMetricGroup {
    /// Latency of pass-through HTTP requests to the pageserver, broken down by pageserver
    /// node id, request name and method. This include both successful and unsuccessful
    /// requests.
    #[metric(metadata = histogram::Thresholds::exponential_buckets(0.1, 2.0))]
    pub(crate) storage_controller_passthrough_request_latency:
        measured::HistogramVec<PageserverRequestLabelGroupSet, 5>,
@@ -74,76 +78,34 @@ pub(crate) struct StorageControllerMetricGroup {
        measured::CounterVec<DatabaseQueryErrorLabelGroupSet>,
    /// Latency of database queries, broken down by operation.
    #[metric(metadata = histogram::Thresholds::exponential_buckets(0.1, 2.0))]
    pub(crate) storage_controller_database_query_latency:
        measured::HistogramVec<DatabaseQueryLatencyLabelGroupSet, 5>,
 }
 impl StorageControllerMetrics {
-    pub(crate) fn encode(&self) -> Bytes {
+    pub(crate) fn encode(&self, neon_metrics: &NeonMetrics) -> Bytes {
        let mut encoder = self.encoder.lock().unwrap();
-        self.metrics_group.collect_into(&mut *encoder);
+        neon_metrics
            .collect_group_into(&mut *encoder)
            .unwrap_or_else(|infallible| match infallible {});
        self.metrics_group
            .collect_group_into(&mut *encoder)
            .unwrap_or_else(|infallible| match infallible {});
        encoder.finish()
    }
 }
 impl Default for StorageControllerMetrics {
    fn default() -> Self {
-        Self {
+        let mut metrics_group = StorageControllerMetricGroup::new();
-            metrics_group: StorageControllerMetricGroup::new(),
+        metrics_group
-            encoder: Mutex::new(measured::text::TextEncoder::new()),
+            .storage_controller_reconcile_complete
-        }
+            .init_all_dense();
    }
 }
 impl StorageControllerMetricGroup {
    pub(crate) fn new() -> Self {
        Self {
-            storage_controller_reconcile_spawn: measured::Counter::new(),
+            metrics_group,
-            storage_controller_reconcile_complete: measured::CounterVec::new(
+            encoder: Mutex::new(measured::text::BufferedTextEncoder::new()),
                ReconcileCompleteLabelGroupSet {
                    status: StaticLabelSet::new(),
                },
            ),
            storage_controller_schedule_optimization: measured::Counter::new(),
            storage_controller_http_request_status: measured::CounterVec::new(
                HttpRequestStatusLabelGroupSet {
                    path: lasso::ThreadedRodeo::new(),
                    method: StaticLabelSet::new(),
                    status: StaticLabelSet::new(),
                },
            ),
            storage_controller_http_request_latency: measured::HistogramVec::new(
                measured::metric::histogram::Thresholds::exponential_buckets(0.1, 2.0),
            ),
            storage_controller_pageserver_request_error: measured::CounterVec::new(
                PageserverRequestLabelGroupSet {
                    pageserver_id: lasso::ThreadedRodeo::new(),
                    path: lasso::ThreadedRodeo::new(),
                    method: StaticLabelSet::new(),
                },
            ),
            storage_controller_pageserver_request_latency: measured::HistogramVec::new(
                measured::metric::histogram::Thresholds::exponential_buckets(0.1, 2.0),
            ),
            storage_controller_passthrough_request_error: measured::CounterVec::new(
                PageserverRequestLabelGroupSet {
                    pageserver_id: lasso::ThreadedRodeo::new(),
                    path: lasso::ThreadedRodeo::new(),
                    method: StaticLabelSet::new(),
                },
            ),
            storage_controller_passthrough_request_latency: measured::HistogramVec::new(
                measured::metric::histogram::Thresholds::exponential_buckets(0.1, 2.0),
            ),
            storage_controller_database_query_error: measured::CounterVec::new(
                DatabaseQueryErrorLabelGroupSet {
                    operation: StaticLabelSet::new(),
                    error_type: StaticLabelSet::new(),
                },
            ),
            storage_controller_database_query_latency: measured::HistogramVec::new(
                measured::metric::histogram::Thresholds::exponential_buckets(0.1, 2.0),
            ),
        }
    }
 }
@@ -157,7 +119,7 @@ pub(crate) struct ReconcileCompleteLabelGroup {
 #[derive(measured::LabelGroup)]
 #[label(set = HttpRequestStatusLabelGroupSet)]
 pub(crate) struct HttpRequestStatusLabelGroup<'a> {
-    #[label(dynamic_with = lasso::ThreadedRodeo)]
+    #[label(dynamic_with = lasso::ThreadedRodeo, default)]
    pub(crate) path: &'a str,
    pub(crate) method: Method,
    pub(crate) status: StatusCode,
@@ -166,40 +128,21 @@ pub(crate) struct HttpRequestStatusLabelGroup<'a> {
 #[derive(measured::LabelGroup)]
 #[label(set = HttpRequestLatencyLabelGroupSet)]
 pub(crate) struct HttpRequestLatencyLabelGroup<'a> {
-    #[label(dynamic_with = lasso::ThreadedRodeo)]
+    #[label(dynamic_with = lasso::ThreadedRodeo, default)]
    pub(crate) path: &'a str,
    pub(crate) method: Method,
 }
 impl Default for HttpRequestLatencyLabelGroupSet {
    fn default() -> Self {
        Self {
            path: lasso::ThreadedRodeo::new(),
            method: StaticLabelSet::new(),
        }
    }
 }
 #[derive(measured::LabelGroup, Clone)]
 #[label(set = PageserverRequestLabelGroupSet)]
 pub(crate) struct PageserverRequestLabelGroup<'a> {
-    #[label(dynamic_with = lasso::ThreadedRodeo)]
+    #[label(dynamic_with = lasso::ThreadedRodeo, default)]
    pub(crate) pageserver_id: &'a str,
-    #[label(dynamic_with = lasso::ThreadedRodeo)]
+    #[label(dynamic_with = lasso::ThreadedRodeo, default)]
    pub(crate) path: &'a str,
    pub(crate) method: Method,
 }
 impl Default for PageserverRequestLabelGroupSet {
    fn default() -> Self {
        Self {
            pageserver_id: lasso::ThreadedRodeo::new(),
            path: lasso::ThreadedRodeo::new(),
            method: StaticLabelSet::new(),
        }
    }
 }
 #[derive(measured::LabelGroup)]
 #[label(set = DatabaseQueryErrorLabelGroupSet)]
 pub(crate) struct DatabaseQueryErrorLabelGroup {
@@ -213,7 +156,7 @@ pub(crate) struct DatabaseQueryLatencyLabelGroup {
    pub(crate) operation: DatabaseOperation,
 }
-#[derive(FixedCardinalityLabel)]
+#[derive(FixedCardinalityLabel, Clone, Copy)]
 pub(crate) enum ReconcileOutcome {
    #[label(rename = "ok")]
    Success,
@@ -221,7 +164,7 @@ pub(crate) enum ReconcileOutcome {
    Cancel,
 }
-#[derive(FixedCardinalityLabel, Clone)]
+#[derive(FixedCardinalityLabel, Copy, Clone)]
 pub(crate) enum Method {
    Get,
    Put,
@@ -246,11 +189,12 @@ impl From<hyper::Method> for Method {
    }
 }
 #[derive(Clone, Copy)]
 pub(crate) struct StatusCode(pub(crate) hyper::http::StatusCode);
 impl LabelValue for StatusCode {
    fn visit<V: measured::label::LabelVisitor>(&self, v: V) -> V::Output {
-        v.write_int(self.0.as_u16() as u64)
+        v.write_int(self.0.as_u16() as i64)
    }
 }
@@ -268,7 +212,7 @@ impl FixedCardinalityLabel for StatusCode {
    }
 }
-#[derive(FixedCardinalityLabel)]
+#[derive(FixedCardinalityLabel, Clone, Copy)]
 pub(crate) enum DatabaseErrorLabel {
    Query,
    Connection,
@@ -79,7 +79,7 @@ pub(crate) enum DatabaseError {
    Logical(String),
 }
-#[derive(measured::FixedCardinalityLabel, Clone)]
+#[derive(measured::FixedCardinalityLabel, Copy, Clone)]
 pub(crate) enum DatabaseOperation {
    InsertNode,
    UpdateNode,
@@ -153,9 +153,7 @@ impl Persistence {
        let latency = &METRICS_REGISTRY
            .metrics_group
            .storage_controller_database_query_latency;
-        let _timer = latency.start_timer(DatabaseQueryLatencyLabelGroup {
+        let _timer = latency.start_timer(DatabaseQueryLatencyLabelGroup { operation: op });
            operation: op.clone(),
        });
        let res = self.with_conn(func).await;
@@ -1,6 +1,7 @@
 import asyncio
 import os
-from typing import Tuple
+import time
 from typing import Optional, Tuple
 import psutil
 import pytest
@@ -20,20 +21,30 @@ ENTRIES_PER_TIMELINE = 10_000
 CHECKPOINT_TIMEOUT_SECONDS = 60
-async def run_worker(env: NeonEnv, tenant_conf, entries: int) -> Tuple[TenantId, TimelineId, Lsn]:
+async def run_worker_for_tenant(
-    tenant, timeline = env.neon_cli.create_tenant(conf=tenant_conf)
+    env: NeonEnv, entries: int, tenant: TenantId, offset: Optional[int] = None
 ) -> Lsn:
    if offset is None:
        offset = 0
    with env.endpoints.create_start("main", tenant_id=tenant) as ep:
        conn = await ep.connect_async()
        try:
            await conn.execute("CREATE TABLE IF NOT EXISTS t(key serial primary key, value text)")
            await conn.execute(
-                f"INSERT INTO t SELECT i, CONCAT('payload_', i) FROM generate_series(0,{entries}) as i"
+                f"INSERT INTO t SELECT i, CONCAT('payload_', i) FROM generate_series({offset},{entries}) as i"
            )
        finally:
            await conn.close(timeout=10)
        last_flush_lsn = Lsn(ep.safe_psql("SELECT pg_current_wal_flush_lsn()")[0][0])
-        return tenant, timeline, last_flush_lsn
+        return last_flush_lsn
 async def run_worker(env: NeonEnv, tenant_conf, entries: int) -> Tuple[TenantId, TimelineId, Lsn]:
    tenant, timeline = env.neon_cli.create_tenant(conf=tenant_conf)
    last_flush_lsn = await run_worker_for_tenant(env, entries, tenant)
    return tenant, timeline, last_flush_lsn
 async def workload(
@@ -89,7 +100,9 @@ def assert_dirty_bytes(env, v):
 def assert_dirty_bytes_nonzero(env):
-    assert get_dirty_bytes(env) > 0
+    dirty_bytes = get_dirty_bytes(env)
    assert dirty_bytes > 0
    return dirty_bytes
@pytest.mark.parametrize("immediate_shutdown", [True, False])
@@ -182,6 +195,31 @@ def test_idle_checkpoints(neon_env_builder: NeonEnvBuilder):
    log.info("Waiting for background checkpoints...")
    wait_until(CHECKPOINT_TIMEOUT_SECONDS * 2, 1, lambda: assert_dirty_bytes(env, 0))  # type: ignore
    # The code below verifies that we do not flush on the first write
    # after an idle period longer than the checkpoint timeout.
    # Sit quietly for longer than the checkpoint timeout
    time.sleep(CHECKPOINT_TIMEOUT_SECONDS + CHECKPOINT_TIMEOUT_SECONDS / 2)
    # Restart the safekeepers and write a bit of extra data into one tenant
    for sk in env.safekeepers:
        sk.start()
    tenant_with_extra_writes = last_flush_lsns[0][0]
    asyncio.run(
        run_worker_for_tenant(env, 5, tenant_with_extra_writes, offset=ENTRIES_PER_TIMELINE)
    )
    dirty_after_write = wait_until(10, 1, lambda: assert_dirty_bytes_nonzero(env))  # type: ignore
    # We shouldn't flush since we've just opened a new layer
    waited_for = 0
    while waited_for < CHECKPOINT_TIMEOUT_SECONDS // 4:
        time.sleep(5)
        waited_for += 5
        assert get_dirty_bytes(env) >= dirty_after_write
@pytest.mark.skipif(
    # We have to use at least ~100MB of data to hit the lowest limit we can configure, which is
@@ -1,5 +1,5 @@
 {
-  "postgres-v16": "3946b2e2ea71d07af092099cb5bcae76a69b90d6",
+  "postgres-v16": "261497dd63ace434045058b1453bcbaaa83f23e5",
-  "postgres-v15": "64b8c7bccc6b77e04795e2d4cf6ad82dc8d987ed",
+  "postgres-v15": "85d809c124a898847a97d66a211f7d5ef4f8e0cb",
-  "postgres-v14": "a7b4c66156bce00afa60e5592d4284ba9e40b4cf"
+  "postgres-v14": "d9149dc59abcbeeb26293707509aef51752db28f"
 }
@@ -37,8 +37,7 @@ futures-io = { version = "0.3" }
 futures-sink = { version = "0.3" }
 futures-util = { version = "0.3", features = ["channel", "io", "sink"] }
 getrandom = { version = "0.2", default-features = false, features = ["std"] }
-hashbrown-582f2526e08bb6a0 = { package = "hashbrown", version = "0.14", features = ["raw"] }
+hashbrown = { version = "0.14", features = ["raw"] }
 hashbrown-594e8ee84c453af0 = { package = "hashbrown", version = "0.13", features = ["raw"] }
 hex = { version = "0.4", features = ["serde"] }
 hmac = { version = "0.12", default-features = false, features = ["reset"] }
 hyper = { version = "0.14", features = ["full"] }
@@ -64,7 +63,7 @@ scopeguard = { version = "1" }
 serde = { version = "1", features = ["alloc", "derive"] }
 serde_json = { version = "1", features = ["raw_value"] }
 sha2 = { version = "0.10", features = ["asm"] }
-smallvec = { version = "1", default-features = false, features = ["write"] }
+smallvec = { version = "1", default-features = false, features = ["const_new", "write"] }
 subtle = { version = "2" }
 time = { version = "0.3", features = ["local-offset", "macros", "serde-well-known"] }
 tokio = { version = "1", features = ["fs", "io-std", "io-util", "macros", "net", "process", "rt-multi-thread", "signal", "test-util"] }
@@ -76,7 +75,6 @@ tonic = { version = "0.9", features = ["tls-roots"] }
 tower = { version = "0.4", default-features = false, features = ["balance", "buffer", "limit", "log", "timeout", "util"] }
 tracing = { version = "0.1", features = ["log"] }
 tracing-core = { version = "0.1" }
 tungstenite = { version = "0.20" }
 url = { version = "2", features = ["serde"] }
 uuid = { version = "1", features = ["serde", "v4", "v7"] }
 zeroize = { version = "1", features = ["derive"] }
@@ -91,7 +89,7 @@ cc = { version = "1", default-features = false, features = ["parallel"] }
 chrono = { version = "0.4", default-features = false, features = ["clock", "serde", "wasmbind"] }
 either = { version = "1" }
 getrandom = { version = "0.2", default-features = false, features = ["std"] }
-hashbrown-582f2526e08bb6a0 = { package = "hashbrown", version = "0.14", features = ["raw"] }
+hashbrown = { version = "0.14", features = ["raw"] }
 indexmap = { version = "1", default-features = false, features = ["std"] }
 itertools = { version = "0.10" }
 libc = { version = "0.2", features = ["extra_traits", "use_std"] }