Use im

Our shutdown procedure for "pageserver init" was buggy. Firstly, it
merely sent the process a SIGKILL, but did not wait for it to actually
exit. Normally, it should exit quickly as SIGKILL cannot be caught or
ignored by the target process, but it's still asynchronous and the
process can still be alive when the kill(2) call returns. Secondly,
"neon_local" removed the PID file after sending SIGKILL, even though the
process was still running. That hid the first problem: if we didn't
remove the PID file, and you start a new pageserver process while the
old one is still running, you would get an error when the new process
tries to lock the PID file.

We've been seeing a lot of "Cannot assign requested address" failures in
the CI lately. Our theory is that when we run "pageserver init"
immediately followed by "pageserver start", the first process is still
running and listening on the port when the second invocation starts up.
This commit hopefully fixes the problem.

It is generally a bad idea for the "neon_local" to remove the PID file
on the child process's behalf. The correct way would be for the server
process to remove the PID file, after it has fully shutdown everything
else. We don't currently have a robust way to ensure that everything has
truly shut down and closed, however.

A simpler way is to simply never remove the PID file. It's not necessary
to remove the PID file for correctness: we cannot rely on the cleanup to
happen anyway, if the server process crashes for example. Because of
that, we already have all the logic in place to deal with a stale PID
file that belonged to a process that already exited. Let's rely on that
on normal shutdown too.

Cargo.lock

@@ -66,6 +66,15 @@ dependencies = [
  "backtrace",
 ]
 
+[[package]]
+name = "archery"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0a8da9bc4c4053ee067669762bcaeea6e241841295a2b6c948312dad6ef4cc02"
+dependencies = [
+ "static_assertions",
+]
+
 [[package]]
 name = "arrayvec"
 version = "0.7.2"
@@ -169,6 +178,333 @@ version = "1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d468802bab17cbc0cc575e9b053f41e72aa36bfa6b7f55e3529ffa43161b97fa"
 
+[[package]]
+name = "aws-config"
+version = "0.51.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "56a636c44c77fa18bdba56126a34d30cfe5538fe88f7d34988fa731fee143ddd"
+dependencies = [
+ "aws-http",
+ "aws-sdk-sso",
+ "aws-sdk-sts",
+ "aws-smithy-async",
+ "aws-smithy-client",
+ "aws-smithy-http",
+ "aws-smithy-http-tower",
+ "aws-smithy-json",
+ "aws-smithy-types",
+ "aws-types",
+ "bytes",
+ "hex",
+ "http",
+ "hyper",
+ "ring",
+ "time 0.3.15",
+ "tokio",
+ "tower",
+ "tracing",
+ "zeroize",
+]
+
+[[package]]
+name = "aws-endpoint"
+version = "0.51.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6ca8f374874f6459aaa88dc861d7f5d834ca1ff97668eae190e97266b5f6c3fb"
+dependencies = [
+ "aws-smithy-http",
+ "aws-smithy-types",
+ "aws-types",
+ "http",
+ "regex",
+ "tracing",
+]
+
+[[package]]
+name = "aws-http"
+version = "0.51.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "78d41e19e779b73463f5f0c21b3aacc995f4ba783ab13a7ae9f5dfb159a551b4"
+dependencies = [
+ "aws-smithy-http",
+ "aws-smithy-types",
+ "aws-types",
+ "bytes",
+ "http",
+ "http-body",
+ "lazy_static",
+ "percent-encoding",
+ "pin-project-lite",
+ "tracing",
+]
+
+[[package]]
+name = "aws-sdk-s3"
+version = "0.21.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a9f08665c8e03aca8cb092ef01e617436ebfa977fddc1240e1b062488ab5d48a"
+dependencies = [
+ "aws-endpoint",
+ "aws-http",
+ "aws-sig-auth",
+ "aws-sigv4",
+ "aws-smithy-async",
+ "aws-smithy-checksums",
+ "aws-smithy-client",
+ "aws-smithy-eventstream",
+ "aws-smithy-http",
+ "aws-smithy-http-tower",
+ "aws-smithy-types",
+ "aws-smithy-xml",
+ "aws-types",
+ "bytes",
+ "bytes-utils",
+ "http",
+ "http-body",
+ "tokio-stream",
+ "tower",
+ "tracing",
+]
+
+[[package]]
+name = "aws-sdk-sso"
+version = "0.21.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "86dcb1cb71aa8763b327542ead410424515cff0cde5b753eedd2917e09c63734"
+dependencies = [
+ "aws-endpoint",
+ "aws-http",
+ "aws-sig-auth",
+ "aws-smithy-async",
+ "aws-smithy-client",
+ "aws-smithy-http",
+ "aws-smithy-http-tower",
+ "aws-smithy-json",
+ "aws-smithy-types",
+ "aws-types",
+ "bytes",
+ "http",
+ "tokio-stream",
+ "tower",
+]
+
+[[package]]
+name = "aws-sdk-sts"
+version = "0.21.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fdfcf584297c666f6b472d5368a78de3bc714b6e0a53d7fbf76c3e347c292ab1"
+dependencies = [
+ "aws-endpoint",
+ "aws-http",
+ "aws-sig-auth",
+ "aws-smithy-async",
+ "aws-smithy-client",
+ "aws-smithy-http",
+ "aws-smithy-http-tower",
+ "aws-smithy-query",
+ "aws-smithy-types",
+ "aws-smithy-xml",
+ "aws-types",
+ "bytes",
+ "http",
+ "tower",
+]
+
+[[package]]
+name = "aws-sig-auth"
+version = "0.51.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "12cbe7b2be9e185c1fbce27fc9c41c66b195b32d89aa099f98768d9544221308"
+dependencies = [
+ "aws-sigv4",
+ "aws-smithy-eventstream",
+ "aws-smithy-http",
+ "aws-types",
+ "http",
+ "tracing",
+]
+
+[[package]]
+name = "aws-sigv4"
+version = "0.51.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "03ff4cff8c4a101962d593ba94e72cd83891aecd423f0c6e3146bff6fb92c9e3"
+dependencies = [
+ "aws-smithy-eventstream",
+ "aws-smithy-http",
+ "bytes",
+ "form_urlencoded",
+ "hex",
+ "http",
+ "once_cell",
+ "percent-encoding",
+ "regex",
+ "ring",
+ "time 0.3.15",
+ "tracing",
+]
+
+[[package]]
+name = "aws-smithy-async"
+version = "0.51.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7b3442b4c5d3fc39891a2e5e625735fba6b24694887d49c6518460fde98247a9"
+dependencies = [
+ "futures-util",
+ "pin-project-lite",
+ "tokio",
+ "tokio-stream",
+]
+
+[[package]]
+name = "aws-smithy-checksums"
+version = "0.51.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cc227e36e346f45298288359f37123e1a92628d1cec6b11b5eb335553278bd9e"
+dependencies = [
+ "aws-smithy-http",
+ "aws-smithy-types",
+ "bytes",
+ "crc32c",
+ "crc32fast",
+ "hex",
+ "http",
+ "http-body",
+ "md-5",
+ "pin-project-lite",
+ "sha1",
+ "sha2",
+ "tracing",
+]
+
+[[package]]
+name = "aws-smithy-client"
+version = "0.51.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ff28d553714f8f54cd921227934fc13a536a1c03f106e56b362fd57e16d450ad"
+dependencies = [
+ "aws-smithy-async",
+ "aws-smithy-http",
+ "aws-smithy-http-tower",
+ "aws-smithy-types",
+ "bytes",
+ "fastrand",
+ "http",
+ "http-body",
+ "hyper",
+ "hyper-rustls",
+ "lazy_static",
+ "pin-project-lite",
+ "tokio",
+ "tower",
+ "tracing",
+]
+
+[[package]]
+name = "aws-smithy-eventstream"
+version = "0.51.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d7ea0df7161ce65b5c8ca6eb709a1a907376fa18226976e41c748ce02ccccf24"
+dependencies = [
+ "aws-smithy-types",
+ "bytes",
+ "crc32fast",
+]
+
+[[package]]
+name = "aws-smithy-http"
+version = "0.51.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bf58ed4fefa61dbf038e5421a521cbc2c448ef69deff0ab1d915d8a10eda5664"
+dependencies = [
+ "aws-smithy-eventstream",
+ "aws-smithy-types",
+ "bytes",
+ "bytes-utils",
+ "futures-core",
+ "http",
+ "http-body",
+ "hyper",
+ "once_cell",
+ "percent-encoding",
+ "pin-project-lite",
+ "pin-utils",
+ "tokio",
+ "tokio-util",
+ "tracing",
+]
+
+[[package]]
+name = "aws-smithy-http-tower"
+version = "0.51.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "20c96d7bd35e7cf96aca1134b2f81b1b59ffe493f7c6539c051791cbbf7a42d3"
+dependencies = [
+ "aws-smithy-http",
+ "bytes",
+ "http",
+ "http-body",
+ "pin-project-lite",
+ "tower",
+ "tracing",
+]
+
+[[package]]
+name = "aws-smithy-json"
+version = "0.51.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d8324ba98c8a94187723cc16c37aefa09504646ee65c3d2c3af495bab5ea701b"
+dependencies = [
+ "aws-smithy-types",
+]
+
+[[package]]
+name = "aws-smithy-query"
+version = "0.51.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "83834ed2ff69ea6f6657baf205267dc2c0abe940703503a3e5d60ce23be3d306"
+dependencies = [
+ "aws-smithy-types",
+ "urlencoding",
+]
+
+[[package]]
+name = "aws-smithy-types"
+version = "0.51.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8b02e06ea63498c43bc0217ea4d16605d4e58d85c12fc23f6572ff6d0a840c61"
+dependencies = [
+ "itoa",
+ "num-integer",
+ "ryu",
+ "time 0.3.15",
+]
+
+[[package]]
+name = "aws-smithy-xml"
+version = "0.51.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "246e9f83dd1fdf5d347fa30ae4ad30a9d1d42ce4cd74a93d94afa874646f94cd"
+dependencies = [
+ "xmlparser",
+]
+
+[[package]]
+name = "aws-types"
+version = "0.51.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "05701d32da168b44f7ee63147781aed8723e792cc131cb9b18363b5393f17f70"
+dependencies = [
+ "aws-smithy-async",
+ "aws-smithy-client",
+ "aws-smithy-http",
+ "aws-smithy-types",
+ "http",
+ "rustc_version 0.4.0",
+ "tracing",
+ "zeroize",
+]
+
 [[package]]
 name = "axum"
 version = "0.5.16"
@@ -300,12 +636,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a"
 
 [[package]]
-name = "block-buffer"
-version = "0.9.0"
+name = "bitmaps"
+version = "2.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4152116fd6e9dadb291ae18fc1ec3575ed6d84c29642d97890f4b4a3417297e4"
+checksum = "031043d04099746d8db04daf1fa424b2bc8bd69d92b25962dcde24da39ab64a2"
 dependencies = [
- "generic-array",
+ "typenum",
 ]
 
 [[package]]
@@ -356,6 +692,16 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "bytes-utils"
+version = "0.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e47d3a8076e283f3acd27400535992edb3ba4b5bb72f8891ad8fbe7932a7d4b9"
+dependencies = [
+ "bytes",
+ "either",
+]
+
 [[package]]
 name = "cast"
 version = "0.3.0"
@@ -806,16 +1152,6 @@ dependencies = [
  "typenum",
 ]
 
-[[package]]
-name = "crypto-mac"
-version = "0.11.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b1d1a86f49236c215f271d40892d5fc950490551400b02ef360692c29815c714"
-dependencies = [
- "generic-array",
- "subtle",
-]
-
 [[package]]
 name = "cxx"
 version = "1.0.79"
@@ -924,47 +1260,17 @@ dependencies = [
  "rusticata-macros",
 ]
 
-[[package]]
-name = "digest"
-version = "0.9.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d3dd60d1080a57a05ab032377049e0591415d2b31afd7028356dbf3cc6dcb066"
-dependencies = [
- "generic-array",
-]
-
 [[package]]
 name = "digest"
 version = "0.10.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "adfbc57365a37acbd2ebf2b64d7e69bb766e2fea813521ed536f5d0520dcf86c"
 dependencies = [
- "block-buffer 0.10.3",
+ "block-buffer",
  "crypto-common",
  "subtle",
 ]
 
-[[package]]
-name = "dirs-next"
-version = "2.0.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b98cf8ebf19c3d1b223e151f99a4f9f0690dca41414773390fc824184ac833e1"
-dependencies = [
- "cfg-if",
- "dirs-sys-next",
-]
-
-[[package]]
-name = "dirs-sys-next"
-version = "0.1.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4ebda144c4fe02d1f7ea1a7d9641b6fc6b580adcfa024ae48797ecdeb6825b4d"
-dependencies = [
- "libc",
- "redox_users",
- "winapi",
-]
-
 [[package]]
 name = "displaydoc"
 version = "0.2.3"
@@ -1109,21 +1415,6 @@ version = "1.0.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1"
 
-[[package]]
-name = "foreign-types"
-version = "0.3.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f6f339eb8adc052cd2ca78910fda869aefa38d22d5cb648e6485e4d3fc06f3b1"
-dependencies = [
- "foreign-types-shared",
-]
-
-[[package]]
-name = "foreign-types-shared"
-version = "0.1.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "00b0228411908ca8685dba7fc2cdd70ec9990a6e753e89b6ac91a84c40fbaf4b"
-
 [[package]]
 name = "form_urlencoded"
 version = "1.1.0"
@@ -1388,23 +1679,13 @@ version = "0.3.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7ebdb29d2ea9ed0083cd8cece49bbd968021bd99b0849edb4a9a7ee0fdf6a4e0"
 
-[[package]]
-name = "hmac"
-version = "0.11.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2a2a2320eb7ec0ebe8da8f744d7812d9fc4cb4d09344ac01898dbcb6a20ae69b"
-dependencies = [
- "crypto-mac",
- "digest 0.9.0",
-]
-
 [[package]]
 name = "hmac"
 version = "0.12.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e"
 dependencies = [
- "digest 0.10.5",
+ "digest",
 ]
 
 [[package]]
@@ -1495,7 +1776,9 @@ checksum = "d87c48c02e0dc5e3b849a2041db3029fd066650f8f717c07bf8ed78ccb895cac"
 dependencies = [
  "http",
  "hyper",
+ "log",
  "rustls",
+ "rustls-native-certs",
  "tokio",
  "tokio-rustls",
 ]
@@ -1512,19 +1795,6 @@ dependencies = [
  "tokio-io-timeout",
 ]
 
-[[package]]
-name = "hyper-tls"
-version = "0.5.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d6183ddfa99b85da61a140bea0efc93fdf56ceaa041b37d553518030827f9905"
-dependencies = [
- "bytes",
- "hyper",
- "native-tls",
- "tokio",
- "tokio-native-tls",
-]
-
 [[package]]
 name = "iana-time-zone"
 version = "0.1.51"
@@ -1565,6 +1835,20 @@ dependencies = [
  "unicode-normalization",
 ]
 
+[[package]]
+name = "im"
+version = "15.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d0acd33ff0285af998aaf9b57342af478078f53492322fafc47450e09397e0e9"
+dependencies = [
+ "bitmaps",
+ "rand_core",
+ "rand_xoshiro",
+ "sized-chunks",
+ "typenum",
+ "version_check",
+]
+
 [[package]]
 name = "indexmap"
 version = "1.9.1"
@@ -1765,24 +2049,13 @@ version = "0.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "73cbba799671b762df5a175adf59ce145165747bb891505c43d09aefbbf38beb"
 
-[[package]]
-name = "md-5"
-version = "0.9.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7b5a279bb9607f9f53c22d496eade00d138d1bdcccd07d74650387cf94942a15"
-dependencies = [
- "block-buffer 0.9.0",
- "digest 0.9.0",
- "opaque-debug",
-]
-
 [[package]]
 name = "md-5"
 version = "0.10.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6365506850d44bff6e2fbcb5176cf63650e48bd45ef2fe2665ae1570e0f4b9ca"
 dependencies = [
- "digest 0.10.5",
+ "digest",
 ]
 
 [[package]]
@@ -1873,24 +2146,6 @@ version = "0.8.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e5ce46fe64a9d73be07dcbe690a38ce1b293be448fd8ce1e6c1b8062c9f72c6a"
 
-[[package]]
-name = "native-tls"
-version = "0.2.10"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fd7e2f3618557f980e0b17e8856252eee3c97fa12c54dff0ca290fb6266ca4a9"
-dependencies = [
- "lazy_static",
- "libc",
- "log",
- "openssl",
- "openssl-probe",
- "openssl-sys",
- "schannel",
- "security-framework",
- "security-framework-sys",
- "tempfile",
-]
-
 [[package]]
 name = "nb"
 version = "0.1.3"
@@ -2061,57 +2316,12 @@ version = "11.1.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0ab1bc2a289d34bd04a330323ac98a1b4bc82c9d9fcb1e66b63caa84da26b575"
 
-[[package]]
-name = "opaque-debug"
-version = "0.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "624a8340c38c1b80fd549087862da4ba43e08858af025b236e509b6649fc13d5"
-
-[[package]]
-name = "openssl"
-version = "0.10.42"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "12fc0523e3bd51a692c8850d075d74dc062ccf251c0110668cbd921917118a13"
-dependencies = [
- "bitflags",
- "cfg-if",
- "foreign-types",
- "libc",
- "once_cell",
- "openssl-macros",
- "openssl-sys",
-]
-
-[[package]]
-name = "openssl-macros"
-version = "0.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b501e44f11665960c7e7fcf062c7d96a14ade4aa98116c004b2e37b5be7d736c"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn",
-]
-
 [[package]]
 name = "openssl-probe"
 version = "0.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ff011a302c396a5197692431fc1948019154afc178baf7d8e37367442a4601cf"
 
-[[package]]
-name = "openssl-sys"
-version = "0.9.76"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5230151e44c0f05157effb743e8d517472843121cf9243e8b81393edb5acd9ce"
-dependencies = [
- "autocfg",
- "cc",
- "libc",
- "pkg-config",
- "vcpkg",
-]
-
 [[package]]
 name = "os_str_bytes"
 version = "6.3.0"
@@ -2150,6 +2360,7 @@ dependencies = [
  "humantime",
  "humantime-serde",
  "hyper",
+ "im",
  "itertools",
  "metrics",
  "nix 0.25.0",
@@ -2167,6 +2378,7 @@ dependencies = [
  "rand",
  "regex",
  "remote_storage",
+ "rpds",
  "rstar",
  "scopeguard",
  "serde",
@@ -2333,12 +2545,6 @@ version = "0.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"
 
-[[package]]
-name = "pkg-config"
-version = "0.3.25"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1df8c4ec4b0627e53bdf214615ad287367e482558cf84b109250b37464dc03ae"
-
 [[package]]
 name = "plotters"
 version = "0.3.4"
@@ -2389,12 +2595,12 @@ dependencies = [
  "byteorder",
  "bytes",
  "fallible-iterator",
- "hmac 0.12.1",
+ "hmac",
  "lazy_static",
- "md-5 0.10.5",
+ "md-5",
  "memchr",
  "rand",
- "sha2 0.10.6",
+ "sha2",
  "stringprep",
 ]
 
@@ -2688,7 +2894,7 @@ dependencies = [
  "git-version",
  "hashbrown",
  "hex",
- "hmac 0.12.1",
+ "hmac",
  "hyper",
  "itertools",
  "md5",
@@ -2707,7 +2913,7 @@ dependencies = [
  "scopeguard",
  "serde",
  "serde_json",
- "sha2 0.10.6",
+ "sha2",
  "socket2",
  "thiserror",
  "tokio",
@@ -2781,6 +2987,15 @@ dependencies = [
  "rand_core",
 ]
 
+[[package]]
+name = "rand_xoshiro"
+version = "0.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6f97cdb2a36ed4183de61b2f824cc45c9f1037f28afe0a322e9fff4c108b5aaa"
+dependencies = [
+ "rand_core",
+]
+
 [[package]]
 name = "rayon"
 version = "1.5.3"
@@ -2826,17 +3041,6 @@ dependencies = [
  "bitflags",
 ]
 
-[[package]]
-name = "redox_users"
-version = "0.4.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b033d837a7cf162d7993aded9304e30a83213c648b6e389db233191f891e5c2b"
-dependencies = [
- "getrandom",
- "redox_syscall",
- "thiserror",
-]
-
 [[package]]
 name = "regex"
 version = "1.6.0"
@@ -2869,10 +3073,13 @@ version = "0.1.0"
 dependencies = [
  "anyhow",
  "async-trait",
+ "aws-config",
+ "aws-sdk-s3",
+ "aws-smithy-http",
+ "aws-types",
+ "hyper",
  "metrics",
  "once_cell",
- "rusoto_core",
- "rusoto_s3",
  "serde",
  "serde_json",
  "tempfile",
@@ -2990,6 +3197,15 @@ dependencies = [
  "regex",
 ]
 
+[[package]]
+name = "rpds"
+version = "0.12.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "66262ea963eff99163e6b741fbc3417a52cc13074728c1047e9911789df9b000"
+dependencies = [
+ "archery",
+]
+
 [[package]]
 name = "rstar"
 version = "0.9.3"
@@ -3026,88 +3242,6 @@ dependencies = [
  "syn",
 ]
 
-[[package]]
-name = "rusoto_core"
-version = "0.48.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1db30db44ea73551326269adcf7a2169428a054f14faf9e1768f2163494f2fa2"
-dependencies = [
- "async-trait",
- "base64",
- "bytes",
- "crc32fast",
- "futures",
- "http",
- "hyper",
- "hyper-tls",
- "lazy_static",
- "log",
- "rusoto_credential",
- "rusoto_signature",
- "rustc_version 0.4.0",
- "serde",
- "serde_json",
- "tokio",
- "xml-rs",
-]
-
-[[package]]
-name = "rusoto_credential"
-version = "0.48.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ee0a6c13db5aad6047b6a44ef023dbbc21a056b6dab5be3b79ce4283d5c02d05"
-dependencies = [
- "async-trait",
- "chrono",
- "dirs-next",
- "futures",
- "hyper",
- "serde",
- "serde_json",
- "shlex",
- "tokio",
- "zeroize",
-]
-
-[[package]]
-name = "rusoto_s3"
-version = "0.48.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7aae4677183411f6b0b412d66194ef5403293917d66e70ab118f07cc24c5b14d"
-dependencies = [
- "async-trait",
- "bytes",
- "futures",
- "rusoto_core",
- "xml-rs",
-]
-
-[[package]]
-name = "rusoto_signature"
-version = "0.48.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a5ae95491c8b4847931e291b151127eccd6ff8ca13f33603eb3d0035ecb05272"
-dependencies = [
- "base64",
- "bytes",
- "chrono",
- "digest 0.9.0",
- "futures",
- "hex",
- "hmac 0.11.0",
- "http",
- "hyper",
- "log",
- "md-5 0.9.1",
- "percent-encoding",
- "pin-project-lite",
- "rusoto_credential",
- "rustc_version 0.4.0",
- "serde",
- "sha2 0.9.9",
- "tokio",
-]
-
 [[package]]
 name = "rustc-demangle"
 version = "0.1.21"
@@ -3159,6 +3293,18 @@ dependencies = [
  "webpki",
 ]
 
+[[package]]
+name = "rustls-native-certs"
+version = "0.6.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0167bac7a9f490495f3c33013e7722b53cb087ecbe082fb0c6387c96f634ea50"
+dependencies = [
+ "openssl-probe",
+ "rustls-pemfile",
+ "schannel",
+ "security-framework",
+]
+
 [[package]]
 name = "rustls-pemfile"
 version = "1.0.1"
@@ -3400,16 +3546,14 @@ dependencies = [
 ]
 
 [[package]]
-name = "sha2"
-version = "0.9.9"
+name = "sha1"
+version = "0.10.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4d58a1e1bf39749807d89cf2d98ac2dfa0ff1cb3faa38fbb64dd88ac8013d800"
+checksum = "f04293dc80c3993519f2d7f6f511707ee7094fe0c6d3406feb330cdb3540eba3"
 dependencies = [
- "block-buffer 0.9.0",
  "cfg-if",
  "cpufeatures",
- "digest 0.9.0",
- "opaque-debug",
+ "digest",
 ]
 
 [[package]]
@@ -3420,7 +3564,7 @@ checksum = "82e6b795fe2e3b1e845bafcb27aa35405c4d47cdfc92af5fc8d3002f76cebdc0"
 dependencies = [
  "cfg-if",
  "cpufeatures",
- "digest 0.10.5",
+ "digest",
 ]
 
 [[package]]
@@ -3486,6 +3630,16 @@ version = "0.3.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7bd3e3206899af3f8b12af284fafc038cc1dc2b41d1b89dd17297221c5d225de"
 
+[[package]]
+name = "sized-chunks"
+version = "0.6.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "16d69225bde7a69b235da73377861095455d298f2b970996eec25ddbb42b3d1e"
+dependencies = [
+ "bitmaps",
+ "typenum",
+]
+
 [[package]]
 name = "slab"
 version = "0.4.7"
@@ -3532,6 +3686,12 @@ version = "1.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a8f112729512f8e442d81f95a8a7ddf2b7c6b8a1a6f509a95864142b30cab2d3"
 
+[[package]]
+name = "static_assertions"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f"
+
 [[package]]
 name = "storage_broker"
 version = "0.1.0"
@@ -3835,16 +3995,6 @@ dependencies = [
  "syn",
 ]
 
-[[package]]
-name = "tokio-native-tls"
-version = "0.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f7d995660bd2b7f8c1568414c1126076c13fbb725c40112dc0120b78eb9b717b"
-dependencies = [
- "native-tls",
- "tokio",
-]
-
 [[package]]
 name = "tokio-postgres"
 version = "0.7.6"
@@ -4227,6 +4377,12 @@ dependencies = [
  "percent-encoding",
 ]
 
+[[package]]
+name = "urlencoding"
+version = "2.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e8db7427f936968176eaa7cdf81b7f98b980b18495ec28f1b5791ac3bfe3eea9"
+
 [[package]]
 name = "utils"
 version = "0.1.0"
@@ -4294,12 +4450,6 @@ version = "0.1.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "77439c1b53d2303b20d9459b1ade71a83c716e3f9c34f3228c00e6f185d6c002"
 
-[[package]]
-name = "vcpkg"
-version = "0.2.15"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426"
-
 [[package]]
 name = "version_check"
 version = "0.9.4"
@@ -4565,7 +4715,6 @@ dependencies = [
  "ahash",
  "anyhow",
  "bytes",
- "chrono",
  "clap 4.0.15",
  "crossbeam-utils",
  "either",
@@ -4595,6 +4744,7 @@ dependencies = [
  "time 0.3.15",
  "tokio",
  "tokio-util",
+ "tower",
  "tracing",
  "tracing-core",
 ]
@@ -4627,10 +4777,10 @@ dependencies = [
 ]
 
 [[package]]
-name = "xml-rs"
-version = "0.8.4"
+name = "xmlparser"
+version = "0.13.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d2d7d3948613f75c98fd9328cfdcc45acc4d360655289d0a7d4ec931392200a3"
+checksum = "4d25c75bf9ea12c4040a97f829154768bbbce366287e2dc044af160cd79a13fd"
 
 [[package]]
 name = "yasna"

control_plane/src/background_process.rs

@@ -147,6 +147,22 @@ where
     anyhow::bail!("{process_name} did not start in {RETRY_UNTIL_SECS} seconds");
 }
 
+/// Send SIGTERM to child process
+pub fn send_stop_child_process(child: &std::process::Child) -> anyhow::Result<()> {
+    let pid = child.id();
+    match kill(
+        nix::unistd::Pid::from_raw(pid.try_into().unwrap()),
+        Signal::SIGTERM,
+    ) {
+        Ok(()) => Ok(()),
+        Err(Errno::ESRCH) => {
+            println!("child process with pid {pid} does not exist");
+            Ok(())
+        }
+        Err(e) => anyhow::bail!("Failed to send signal to child process with pid {pid}: {e}"),
+    }
+}
+
 /// Stops the process, using the pid file given. Returns Ok also if the process is already not running.
 pub fn stop_process(immediate: bool, process_name: &str, pid_file: &Path) -> anyhow::Result<()> {
     if !pid_file.exists() {
@@ -179,11 +195,6 @@ pub fn stop_process(immediate: bool, process_name: &str, pid_file: &Path) -> any
         match process_has_stopped(pid) {
             Ok(true) => {
                 println!("\n{process_name} stopped");
-                if let Err(e) = fs::remove_file(pid_file) {
-                    if e.kind() != io::ErrorKind::NotFound {
-                        eprintln!("Failed to remove pid file {pid_file:?} after stopping the process: {e:#}");
-                    }
-                }
                 return Ok(());
             }
             Ok(false) => {

control_plane/src/pageserver.rs

@@ -1,12 +1,12 @@
 use std::collections::HashMap;
-use std::fs::{self, File};
+use std::fs::File;
 use std::io::{BufReader, Write};
 use std::num::NonZeroU64;
 use std::path::{Path, PathBuf};
 use std::process::Child;
 use std::{io, result};
 
-use anyhow::{bail, Context};
+use anyhow::{bail, ensure, Context};
 use pageserver_api::models::{
     TenantConfigRequest, TenantCreateRequest, TenantInfo, TimelineCreateRequest, TimelineInfo,
 };
@@ -168,29 +168,21 @@ impl PageServerNode {
             }
             Err(e) => eprintln!("{e:#}"),
         }
-        match pageserver_process.kill() {
-            Err(e) => {
-                eprintln!(
-                    "Failed to stop pageserver {} process with pid {}: {e:#}",
-                    self.env.pageserver.id,
-                    pageserver_process.id(),
-                )
-            }
-            Ok(()) => {
-                println!(
-                    "Stopped pageserver {} process with pid {}",
-                    self.env.pageserver.id,
-                    pageserver_process.id(),
-                );
-                // cleanup after pageserver startup, since we do not call regular `stop_process` during init
-                let pid_file = self.pid_file();
-                if let Err(e) = fs::remove_file(&pid_file) {
-                    if e.kind() != io::ErrorKind::NotFound {
-                        eprintln!("Failed to remove pid file {pid_file:?} after stopping the process: {e:#}");
-                    }
-                }
-            }
-        }
+        background_process::send_stop_child_process(&pageserver_process)?;
+
+        let exit_code = pageserver_process.wait()?;
+        ensure!(
+            exit_code.success(),
+            format!(
+                "pageserver init failed with exit code {:?}",
+                exit_code.code()
+            )
+        );
+        println!(
+            "Stopped pageserver {} process with pid {}",
+            self.env.pageserver.id,
+            pageserver_process.id(),
+        );
         init_result
     }
 

libs/remote_storage/Cargo.toml

@@ -9,8 +9,11 @@ async-trait = "0.1"
 metrics = { version = "0.1", path = "../metrics" }
 utils = { version = "0.1", path = "../utils" }
 once_cell = "1.13.0"
-rusoto_core = "0.48"
-rusoto_s3 = "0.48"
+aws-smithy-http = "0.51.0"
+aws-types = "0.51.0"
+aws-config = { version = "0.51.0", default-features = false, features=["rustls"] }
+aws-sdk-s3 = "0.21.0"
+hyper = { version = "0.14", features = ["stream"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1"
 tokio = { version = "1.17", features = ["sync", "macros", "fs", "io-util"] }

libs/remote_storage/src/s3_bucket.rs

@@ -4,27 +4,36 @@
 //! allowing multiple api users to independently work with the same S3 bucket, if
 //! their bucket prefixes are both specified and different.
 
+use std::env::var;
 use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use std::time::Duration;
 
 use anyhow::Context;
-use rusoto_core::{
-    credential::{InstanceMetadataProvider, StaticProvider},
-    HttpClient, Region, RusotoError,
+use aws_config::{
+    environment::credentials::EnvironmentVariableCredentialsProvider, imds,
+    imds::credentials::ImdsCredentialsProvider, meta::credentials::provide_credentials_fn,
 };
-use rusoto_s3::{
-    DeleteObjectRequest, GetObjectError, GetObjectRequest, ListObjectsV2Request, PutObjectRequest,
-    S3Client, StreamingBody, S3,
+use aws_sdk_s3::{
+    config::Config,
+    error::{GetObjectError, GetObjectErrorKind},
+    types::{ByteStream, SdkError},
+    Client, Endpoint, Region,
 };
+use aws_smithy_http::body::SdkBody;
+use aws_types::credentials::{CredentialsError, ProvideCredentials};
+use hyper::Body;
 use tokio::{io, sync::Semaphore};
 use tokio_util::io::ReaderStream;
 use tracing::debug;
 
+use super::StorageMetadata;
 use crate::{
     strip_path_prefix, Download, DownloadError, RemoteObjectId, RemoteStorage, S3Config,
     REMOTE_STORAGE_PREFIX_SEPARATOR,
 };
 
-use super::StorageMetadata;
+const DEFAULT_IMDS_TIMEOUT: Duration = Duration::from_secs(10);
 
 pub(super) mod metrics {
     use metrics::{register_int_counter_vec, IntCounterVec};
@@ -116,7 +125,7 @@ fn download_destination(
 /// AWS S3 storage.
 pub struct S3Bucket {
     workdir: PathBuf,
-    client: S3Client,
+    client: Client,
     bucket_name: String,
     prefix_in_bucket: Option<String>,
     // Every request to S3 can be throttled or cancelled, if a certain number of requests per second is exceeded.
@@ -125,6 +134,12 @@ pub struct S3Bucket {
     concurrency_limiter: Semaphore,
 }
 
+#[derive(Default)]
+struct GetObjectRequest {
+    bucket: String,
+    key: String,
+    range: Option<String>,
+}
 impl S3Bucket {
     /// Creates the S3 storage, errors if incorrect AWS S3 configuration provided.
     pub fn new(aws_config: &S3Config, workdir: PathBuf) -> anyhow::Result<Self> {
@@ -132,43 +147,40 @@ impl S3Bucket {
             "Creating s3 remote storage for S3 bucket {}",
             aws_config.bucket_name
         );
-        let region = match aws_config.endpoint.clone() {
-            Some(custom_endpoint) => Region::Custom {
-                name: aws_config.bucket_region.clone(),
-                endpoint: custom_endpoint,
-            },
-            None => aws_config
-                .bucket_region
-                .parse::<Region>()
-                .context("Failed to parse the s3 region from config")?,
-        };
-        let request_dispatcher = HttpClient::new().context("Failed to create S3 http client")?;
+        let mut config_builder = Config::builder()
+            .region(Region::new(aws_config.bucket_region.clone()))
+            .credentials_provider(provide_credentials_fn(|| async {
+                match var("AWS_ACCESS_KEY_ID").is_ok() && var("AWS_SECRET_ACCESS_KEY").is_ok() {
+                    true => {
+                        EnvironmentVariableCredentialsProvider::new()
+                            .provide_credentials()
+                            .await
+                    }
+                    false => {
+                        let imds_client = imds::Client::builder()
+                            .connect_timeout(DEFAULT_IMDS_TIMEOUT)
+                            .read_timeout(DEFAULT_IMDS_TIMEOUT)
+                            .build()
+                            .await
+                            .map_err(CredentialsError::unhandled)?;
+                        ImdsCredentialsProvider::builder()
+                            .imds_client(imds_client)
+                            .build()
+                            .provide_credentials()
+                            .await
+                    }
+                }
+            }));
 
-        let access_key_id = std::env::var("AWS_ACCESS_KEY_ID").ok();
-        let secret_access_key = std::env::var("AWS_SECRET_ACCESS_KEY").ok();
-        // session token is used when authorizing through sso
-        // which is typically the case when testing locally on developer machine
-        let session_token = std::env::var("AWS_SESSION_TOKEN").ok();
-
-        let client = if access_key_id.is_none() && secret_access_key.is_none() {
-            debug!("Using IAM-based AWS access");
-            S3Client::new_with(request_dispatcher, InstanceMetadataProvider::new(), region)
-        } else {
-            debug!(
-                "Using credentials-based AWS access. Session token is set: {}",
-                session_token.is_some()
+        if let Some(custom_endpoint) = aws_config.endpoint.clone() {
+            let endpoint = Endpoint::immutable(
+                custom_endpoint
+                    .parse()
+                    .expect("Failed to parse S3 custom endpoint"),
             );
-            S3Client::new_with(
-                request_dispatcher,
-                StaticProvider::new(
-                    access_key_id.unwrap_or_default(),
-                    secret_access_key.unwrap_or_default(),
-                    session_token,
-                    None,
-                ),
-                region,
-            )
-        };
+            config_builder.set_endpoint_resolver(Some(Arc::new(endpoint)));
+        }
+        let client = Client::from_conf(config_builder.build());
 
         let prefix_in_bucket = aws_config.prefix_in_bucket.as_deref().map(|prefix| {
             let mut prefix = prefix;
@@ -182,7 +194,6 @@ impl S3Bucket {
             }
             prefix
         });
-
         Ok(Self {
             client,
             workdir,
@@ -202,20 +213,33 @@ impl S3Bucket {
 
         metrics::inc_get_object();
 
-        match self.client.get_object(request).await {
-            Ok(object_output) => match object_output.body {
-                None => {
-                    metrics::inc_get_object_fail();
-                    Err(DownloadError::Other(anyhow::anyhow!(
-                        "Got no body for the S3 object given"
-                    )))
-                }
-                Some(body) => Ok(Download {
-                    metadata: object_output.metadata.map(StorageMetadata),
-                    download_stream: Box::pin(io::BufReader::new(body.into_async_read())),
-                }),
-            },
-            Err(RusotoError::Service(GetObjectError::NoSuchKey(_))) => Err(DownloadError::NotFound),
+        let get_object = self
+            .client
+            .get_object()
+            .bucket(request.bucket)
+            .key(request.key)
+            .set_range(request.range)
+            .send()
+            .await;
+
+        match get_object {
+            Ok(object_output) => {
+                let metadata = object_output.metadata().cloned().map(StorageMetadata);
+                Ok(Download {
+                    metadata,
+                    download_stream: Box::pin(io::BufReader::new(
+                        object_output.body.into_async_read(),
+                    )),
+                })
+            }
+            Err(SdkError::ServiceError {
+                err:
+                    GetObjectError {
+                        kind: GetObjectErrorKind::NoSuchKey(..),
+                        ..
+                    },
+                ..
+            }) => Err(DownloadError::NotFound),
             Err(e) => {
                 metrics::inc_get_object_fail();
                 Err(DownloadError::Other(anyhow::anyhow!(
@@ -261,12 +285,11 @@ impl RemoteStorage for S3Bucket {
 
             let fetch_response = self
                 .client
-                .list_objects_v2(ListObjectsV2Request {
-                    bucket: self.bucket_name.clone(),
-                    prefix: self.prefix_in_bucket.clone(),
-                    continuation_token,
-                    ..ListObjectsV2Request::default()
-                })
+                .list_objects_v2()
+                .bucket(self.bucket_name.clone())
+                .set_prefix(self.prefix_in_bucket.clone())
+                .set_continuation_token(continuation_token)
+                .send()
                 .await
                 .map_err(|e| {
                     metrics::inc_list_objects_fail();
@@ -322,13 +345,12 @@ impl RemoteStorage for S3Bucket {
 
             let fetch_response = self
                 .client
-                .list_objects_v2(ListObjectsV2Request {
-                    bucket: self.bucket_name.clone(),
-                    prefix: list_prefix.clone(),
-                    continuation_token,
-                    delimiter: Some(REMOTE_STORAGE_PREFIX_SEPARATOR.to_string()),
-                    ..ListObjectsV2Request::default()
-                })
+                .list_objects_v2()
+                .bucket(self.bucket_name.clone())
+                .set_prefix(list_prefix.clone())
+                .set_continuation_token(continuation_token)
+                .delimiter(REMOTE_STORAGE_PREFIX_SEPARATOR.to_string())
+                .send()
                 .await
                 .map_err(|e| {
                     metrics::inc_list_objects_fail();
@@ -366,17 +388,18 @@ impl RemoteStorage for S3Bucket {
             .context("Concurrency limiter semaphore got closed during S3 upload")?;
 
         metrics::inc_put_object();
+
+        let body = Body::wrap_stream(ReaderStream::new(from));
+        let bytes_stream = ByteStream::new(SdkBody::from(body));
+
         self.client
-            .put_object(PutObjectRequest {
-                body: Some(StreamingBody::new_with_size(
-                    ReaderStream::new(from),
-                    from_size_bytes,
-                )),
-                bucket: self.bucket_name.clone(),
-                key: to.0.to_owned(),
-                metadata: metadata.map(|m| m.0),
-                ..PutObjectRequest::default()
-            })
+            .put_object()
+            .bucket(self.bucket_name.clone())
+            .key(to.0.to_owned())
+            .set_metadata(metadata.map(|m| m.0))
+            .content_length(from_size_bytes.try_into()?)
+            .body(bytes_stream)
+            .send()
             .await
             .map_err(|e| {
                 metrics::inc_put_object_fail();
@@ -412,7 +435,6 @@ impl RemoteStorage for S3Bucket {
             bucket: self.bucket_name.clone(),
             key: from.0.to_owned(),
             range,
-            ..GetObjectRequest::default()
         })
         .await
     }
@@ -427,11 +449,10 @@ impl RemoteStorage for S3Bucket {
         metrics::inc_delete_object();
 
         self.client
-            .delete_object(DeleteObjectRequest {
-                bucket: self.bucket_name.clone(),
-                key: remote_object_id.0.to_owned(),
-                ..DeleteObjectRequest::default()
-            })
+            .delete_object()
+            .bucket(self.bucket_name.clone())
+            .key(remote_object_id.0.to_owned())
+            .send()
             .await
             .map_err(|e| {
                 metrics::inc_delete_object_fail();
@@ -600,7 +621,7 @@ mod tests {
     fn dummy_storage(workdir: PathBuf) -> S3Bucket {
         S3Bucket {
             workdir,
-            client: S3Client::new("us-east-1".parse().unwrap()),
+            client: Client::new(&aws_config::SdkConfig::builder().build()),
             bucket_name: "dummy-bucket".to_string(),
             prefix_in_bucket: Some("dummy_prefix/".to_string()),
             concurrency_limiter: Semaphore::new(1),

pageserver/Cargo.toml

@@ -69,6 +69,8 @@ remote_storage = { path = "../libs/remote_storage" }
 tenant_size_model = { path = "../libs/tenant_size_model" }
 utils = { path = "../libs/utils" }
 workspace_hack = { version = "0.1", path = "../workspace_hack" }
+rpds = "0.12.0"
+im = "15.1.0"
 
 [dev-dependencies]
 criterion = "0.4"

pageserver/benches/README.md

@@ -0,0 +1,12 @@
+## Pageserver Benchmarks
+
+# How to run
+
+To run all benchmarks:
+`cargo bench`
+
+To run a specific file:
+`cargo bench --bench bench_layer_map`
+
+To run a specific function:
+`cargo bench --bench bench_layer_map -- real_map_uniform_queries`

pageserver/src/tenant.rs

@@ -73,6 +73,7 @@ use utils::{
 
 mod blob_io;
 pub mod block_io;
+pub mod bst_layer_map;
 mod delta_layer;
 mod disk_btree;
 pub(crate) mod ephemeral_file;

pageserver/src/tenant/bst_layer_map.rs

@@ -0,0 +1,407 @@
+use std::collections::BTreeMap;
+use std::ops::Range;
+
+// TODO drop rpds. So far `im` looks 30% faster.
+use rpds::RedBlackTreeMapSync;
+use im::OrdMap;
+
+/// Layer map implemented using persistent/immutable binary search tree.
+/// It supports historical queries, but no retroactive inserts. For that
+/// see RetroactiveLayerMap.
+///
+/// Layer type is abstracted as Value to make unit testing easier.
+pub struct PersistentLayerMap<Value> {
+    /// Mapping key to the latest layer (if any) until the next key.
+    /// We use the Sync version of the map because we want Self to
+    /// be Sync.
+    ///
+    /// TODO Separate Head into its own struct LatestLayerMap
+    /// TODO Merge historic with retroactive, into HistoricLayerMap
+    /// TODO Maintain a pair of heads, one for images, one for deltas.
+    ///      This way we can query both of them with one BTreeMap query.
+    head: OrdMap<i128, Option<(u64, Value)>>,
+
+    /// All previous states of `self.head`
+    ///
+    /// TODO: Sorted Vec + binary search could be slightly faster.
+    historic: BTreeMap<u64, OrdMap<i128, Option<(u64, Value)>>>,
+}
+
+impl<Value: std::fmt::Debug> std::fmt::Debug for PersistentLayerMap<Value> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let head_vec: Vec<_> = self.head.iter().collect();
+        write!(f, "PersistentLayerMap: head: {:?}", head_vec)
+    }
+}
+
+impl<T: Clone> Default for PersistentLayerMap<T> {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl<Value: Clone> PersistentLayerMap<Value> {
+    pub fn new() -> Self {
+        Self {
+            head: OrdMap::default(),
+            historic: BTreeMap::default(),
+        }
+    }
+
+    /// Helper function to subdivide the key range without changing any values
+    fn add_node(self: &mut Self, key: i128) {
+        let value = match self.head.range(0..=key).last() {
+            Some((_, Some(v))) => Some(v.clone()),
+            Some((_, None)) => None,
+            None => None,
+        };
+        self.head.insert(key, value);
+    }
+
+    pub fn insert(self: &mut Self, key: Range<i128>, lsn: Range<u64>, value: Value) {
+        // It's only a persistent map, not a retroactive one
+        if let Some(last_entry) = self.historic.iter().rev().next() {
+            let last_lsn = last_entry.0;
+            if lsn.start == *last_lsn {
+                // TODO there are edge cases to take care of
+            }
+            if lsn.start < *last_lsn {
+                panic!("unexpected retroactive insert");
+            }
+        }
+
+        // NOTE The order of the following lines is important!!
+
+        // Add nodes at endpoints
+        self.add_node(key.start);
+        self.add_node(key.end);
+
+        // Raise the height where necessary
+        //
+        // NOTE This loop is worst case O(N), but amortized O(log N) in the special
+        // case when rectangles have no height. In practice I don't think we'll see
+        // the kind of layer intersections needed to trigger O(N) behavior. If we
+        // do it can be fixed using lazy propagation.
+        let mut to_update = Vec::new();
+        let mut to_remove = Vec::new();
+        let mut prev_covered = false;
+        for (k, node) in self.head.range(key.clone()) {
+            let needs_cover = match node {
+                None => true,
+                Some((h, _)) => h < &lsn.end,
+            };
+            if needs_cover {
+                match prev_covered {
+                    true => to_remove.push(k.clone()),
+                    false => to_update.push(k.clone()),
+                }
+            }
+            prev_covered = needs_cover;
+        }
+        if !prev_covered {
+            to_remove.push(key.end);
+        }
+        for k in to_update {
+            self.head
+                .insert(k.clone(), Some((lsn.end.clone(), value.clone())));
+        }
+        for k in to_remove {
+            self.head.remove(&k);
+        }
+
+        // Remember history. Clone is O(1)
+        self.historic.insert(lsn.start, self.head.clone());
+    }
+
+    pub fn query(self: &Self, key: i128, lsn: u64) -> Option<Value> {
+        let version = self.historic.range(0..=lsn).rev().next()?.1;
+        version
+            .get_prev(&key)?
+            // .range(0..=key).rev().next()?
+            // NOTE The canonical way to do this in other crates is
+            //      `.range(0..=key).rev.next()` and `im` supports this
+            //      API but it's 2x slower than `.get_prev(&key)`.
+            .1
+            .as_ref()
+            .map(|(_, v)| v.clone())
+    }
+
+    pub fn trim(self: &mut Self, begin: &u64) {
+        self.historic.split_off(begin);
+        self.head = self
+            .historic
+            .iter()
+            .rev()
+            .next()
+            .map(|(_, v)| v.clone())
+            .unwrap_or_default();
+    }
+}
+
+/// Basic test for the immutable bst library, just to show usage.
+#[test]
+fn test_immutable_bst_dependency() {
+    let map = RedBlackTreeMapSync::<i32, i32>::default();
+
+    let mut v1 = map.clone();
+    let v2 = map.insert(1, 5);
+
+    // We can query current and past versions of key 1
+    assert_eq!(v1.get(&1), None);
+    assert_eq!(v2.get(&1), Some(&5));
+
+    // We can mutate old state, but it creates a branch.
+    // It doesn't retroactively change future versions.
+    v1.insert_mut(2, 6);
+    assert_eq!(v1.get(&2), Some(&6));
+    assert_eq!(v2.get(&2), None);
+}
+
+/// This is the most basic test that demonstrates intended usage.
+/// All layers in this test have height 1.
+#[test]
+fn test_persistent_simple() {
+    let mut map = PersistentLayerMap::<String>::new();
+    map.insert(0..5, 100..101, "Layer 1".to_string());
+    map.insert(3..9, 110..111, "Layer 2".to_string());
+    map.insert(5..6, 120..121, "Layer 3".to_string());
+
+    // After Layer 1 insertion
+    assert_eq!(map.query(1, 105), Some("Layer 1".to_string()));
+    assert_eq!(map.query(4, 105), Some("Layer 1".to_string()));
+
+    // After Layer 2 insertion
+    assert_eq!(map.query(4, 115), Some("Layer 2".to_string()));
+    assert_eq!(map.query(8, 115), Some("Layer 2".to_string()));
+    assert_eq!(map.query(11, 115), None);
+
+    // After Layer 3 insertion
+    assert_eq!(map.query(4, 125), Some("Layer 2".to_string()));
+    assert_eq!(map.query(5, 125), Some("Layer 3".to_string()));
+    assert_eq!(map.query(7, 125), Some("Layer 2".to_string()));
+}
+
+/// Cover simple off-by-one edge cases
+#[test]
+fn test_off_by_one() {
+    let mut map = PersistentLayerMap::<String>::new();
+    map.insert(3..5, 100..110, "Layer 1".to_string());
+
+    // Check different LSNs
+    assert_eq!(map.query(4, 99), None);
+    assert_eq!(map.query(4, 100), Some("Layer 1".to_string()));
+
+    // Check different keys
+    assert_eq!(map.query(2, 105), None);
+    assert_eq!(map.query(3, 105), Some("Layer 1".to_string()));
+    assert_eq!(map.query(4, 105), Some("Layer 1".to_string()));
+    assert_eq!(map.query(5, 105), None);
+}
+
+/// Cover edge cases where layers begin or end on the same key
+#[test]
+fn test_key_collision() {
+    let mut map = PersistentLayerMap::<String>::new();
+
+    map.insert(3..5, 100..110, "Layer 10".to_string());
+    map.insert(5..8, 100..110, "Layer 11".to_string());
+
+    map.insert(3..4, 200..210, "Layer 20".to_string());
+
+    // Check after layer 11
+    assert_eq!(map.query(2, 105), None);
+    assert_eq!(map.query(3, 105), Some("Layer 10".to_string()));
+    assert_eq!(map.query(5, 105), Some("Layer 11".to_string()));
+    assert_eq!(map.query(7, 105), Some("Layer 11".to_string()));
+    assert_eq!(map.query(8, 105), None);
+
+    // Check after layer 20
+    assert_eq!(map.query(2, 205), None);
+    assert_eq!(map.query(3, 205), Some("Layer 20".to_string()));
+    assert_eq!(map.query(5, 205), Some("Layer 11".to_string()));
+    assert_eq!(map.query(7, 205), Some("Layer 11".to_string()));
+    assert_eq!(map.query(8, 205), None);
+}
+
+/// Test when rectangles have nontrivial height and possibly overlap
+#[test]
+fn test_persistent_overlapping() {
+    let mut map = PersistentLayerMap::<String>::new();
+
+    // Add 3 key-disjoint layers with varying LSN ranges
+    map.insert(1..2, 100..200, "Layer 1".to_string());
+    map.insert(4..5, 110..200, "Layer 2".to_string());
+    map.insert(7..8, 120..300, "Layer 3".to_string());
+
+    // Add wide and short layer
+    map.insert(0..9, 130..199, "Layer 4".to_string());
+
+    // Add wide layer taller than some
+    map.insert(0..9, 140..201, "Layer 5".to_string());
+
+    // Add wide layer taller than all
+    map.insert(0..9, 150..301, "Layer 6".to_string());
+
+    // After layer 4 insertion
+    assert_eq!(map.query(0, 135), Some("Layer 4".to_string()));
+    assert_eq!(map.query(1, 135), Some("Layer 1".to_string()));
+    assert_eq!(map.query(2, 135), Some("Layer 4".to_string()));
+    assert_eq!(map.query(4, 135), Some("Layer 2".to_string()));
+    assert_eq!(map.query(5, 135), Some("Layer 4".to_string()));
+    assert_eq!(map.query(7, 135), Some("Layer 3".to_string()));
+    assert_eq!(map.query(8, 135), Some("Layer 4".to_string()));
+
+    // After layer 5 insertion
+    assert_eq!(map.query(0, 145), Some("Layer 5".to_string()));
+    assert_eq!(map.query(1, 145), Some("Layer 5".to_string()));
+    assert_eq!(map.query(2, 145), Some("Layer 5".to_string()));
+    assert_eq!(map.query(4, 145), Some("Layer 5".to_string()));
+    assert_eq!(map.query(5, 145), Some("Layer 5".to_string()));
+    assert_eq!(map.query(7, 145), Some("Layer 3".to_string()));
+    assert_eq!(map.query(8, 145), Some("Layer 5".to_string()));
+
+    // After layer 6 insertion
+    assert_eq!(map.query(0, 155), Some("Layer 6".to_string()));
+    assert_eq!(map.query(1, 155), Some("Layer 6".to_string()));
+    assert_eq!(map.query(2, 155), Some("Layer 6".to_string()));
+    assert_eq!(map.query(4, 155), Some("Layer 6".to_string()));
+    assert_eq!(map.query(5, 155), Some("Layer 6".to_string()));
+    assert_eq!(map.query(7, 155), Some("Layer 6".to_string()));
+    assert_eq!(map.query(8, 155), Some("Layer 6".to_string()));
+}
+
+/// Layer map that supports:
+/// - efficient historical queries
+/// - efficient append only updates
+/// - tombstones and similar methods for non-latest updates
+/// - compaction/rebuilding to remove tombstones
+///
+/// See this for better retroactive techniques we can try
+/// https://www.youtube.com/watch?v=WqCWghETNDc&t=581s
+///
+/// Layer type is abstracted as Value to make unit testing easier.
+pub struct RetroactiveLayerMap<Value> {
+    /// A persistent layer map that we rebuild when we need to retroactively update
+    map: PersistentLayerMap<Value>,
+
+    /// We buffer insertion into the PersistentLayerMap to decrease the number of rebuilds.
+    /// A value of None means we want to delete this item.
+    buffer: BTreeMap<(u64, u64, i128, i128), Option<Value>>,
+
+    /// All current layers. This is not used for search. Only to make rebuilds easier.
+    layers: BTreeMap<(u64, u64, i128, i128), Value>,
+}
+
+impl<Value: std::fmt::Debug> std::fmt::Debug for RetroactiveLayerMap<Value> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "RetroactiveLayerMap: head: {:?}", self.map)
+    }
+}
+
+impl<T: Clone> Default for RetroactiveLayerMap<T> {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl<Value: Clone> RetroactiveLayerMap<Value> {
+    pub fn new() -> Self {
+        Self {
+            map: PersistentLayerMap::<Value>::new(),
+            buffer: BTreeMap::new(),
+            layers: BTreeMap::new(),
+        }
+    }
+
+    pub fn insert(self: &mut Self, key: Range<i128>, lsn: Range<u64>, value: Value) {
+        self.buffer.insert(
+            (lsn.start, lsn.end, key.start, key.end),
+            Some(value.clone()),
+        );
+    }
+
+    pub fn remove(self: &mut Self, key: Range<i128>, lsn: Range<u64>) {
+        self.buffer
+            .insert((lsn.start, lsn.end, key.start, key.end), None);
+    }
+
+    pub fn rebuild(self: &mut Self) {
+        // Find the first LSN that needs to be rebuilt
+        let rebuild_since: u64 = match self.buffer.iter().next() {
+            Some(((lsn_start, _, _, _), _)) => lsn_start.clone(),
+            None => return, // No need to rebuild if buffer is empty
+        };
+
+        // Apply buffered updates to self.layers
+        self.buffer.retain(|rect, layer| {
+            match layer {
+                Some(l) => {
+                    let existing = self.layers.insert(rect.clone(), l.clone());
+                    if existing.is_some() {
+                        panic!("can't overwrite layer");
+                    }
+                }
+                None => {
+                    let existing = self.layers.remove(rect);
+                    if existing.is_none() {
+                        panic!("invalid layer deletion");
+                    }
+                }
+            };
+            false
+        });
+
+        // Rebuild
+        self.map.trim(&rebuild_since);
+        for ((lsn_start, lsn_end, key_start, key_end), layer) in
+            self.layers.range((rebuild_since, 0, 0, 0)..)
+        {
+            self.map
+                .insert(*key_start..*key_end, *lsn_start..*lsn_end, layer.clone());
+        }
+    }
+
+    pub fn clear(self: &mut Self) {
+        self.map.trim(&0);
+    }
+
+    pub fn query(self: &Self, key: i128, lsn: u64) -> Option<Value> {
+        if !self.buffer.is_empty() {
+            panic!("rebuild pls")
+        }
+
+        self.map.query(key, lsn)
+    }
+}
+
+#[test]
+fn test_retroactive_simple() {
+    let mut map = RetroactiveLayerMap::new();
+
+    // Append some images in increasing LSN order
+    map.insert(0..5, 100..101, "Image 1".to_string());
+    map.insert(3..9, 110..111, "Image 2".to_string());
+    map.insert(4..6, 120..121, "Image 3".to_string());
+    map.insert(8..9, 120..121, "Image 4".to_string());
+
+    // Add a delta layer out of order
+    map.insert(2..5, 105..106, "Delta 1".to_string());
+
+    // Rebuild so we can start querying
+    map.rebuild();
+
+    // Query key 4
+    assert_eq!(map.query(4, 90), None);
+    assert_eq!(map.query(4, 102), Some("Image 1".to_string()));
+    assert_eq!(map.query(4, 107), Some("Delta 1".to_string()));
+    assert_eq!(map.query(4, 115), Some("Image 2".to_string()));
+    assert_eq!(map.query(4, 125), Some("Image 3".to_string()));
+
+    // Remove Image 3
+    map.remove(4..6, 120..121);
+    map.rebuild();
+
+    // Check deletion worked
+    assert_eq!(map.query(4, 125), Some("Image 2".to_string()));
+    assert_eq!(map.query(8, 125), Some("Image 4".to_string()));
+}

pageserver/src/tenant/layer_map.rs

@@ -28,6 +28,8 @@ use std::sync::Arc;
 use tracing::*;
 use utils::lsn::Lsn;
 
+use super::bst_layer_map::RetroactiveLayerMap;
+
 ///
 /// LayerMap tracks what layers exist on a timeline.
 ///
@@ -55,6 +57,11 @@ pub struct LayerMap {
     /// All the historic layers are kept here
     historic_layers: RTree<LayerRTreeObject>,
 
+    /// HACK I'm experimenting with a new index to reaplace the RTree. If this
+    ///      works out I'll clean up the struct later.
+    index: RetroactiveLayerMap<Arc<dyn Layer>>,
+    images: RetroactiveLayerMap<Arc<dyn Layer>>,
+
     /// L0 layers have key range Key::MIN..Key::MAX, and locating them using R-Tree search is very inefficient.
     /// So L0 layers are held in l0_delta_layers vector, in addition to the R-tree.
     l0_delta_layers: Vec<Arc<dyn Layer>>,
@@ -241,6 +248,65 @@ impl LayerMap {
     /// layer.
     ///
     pub fn search(&self, key: Key, end_lsn: Lsn) -> Result<Option<SearchResult>> {
+        let old = self.search_old(key, end_lsn)?;
+        let new = self.search_new(key, end_lsn)?;
+        match (&old, &new) {
+            (None, None) => {}
+            (None, Some(_)) => panic!("returned Some, expected None"),
+            (Some(_), None) => panic!("returned None, expected Some"),
+            (Some(old), Some(new)) => {
+                // TODO be more verbose and flexible
+                let context = format!("query: key {}, end_lsn: {}", key, end_lsn);
+                assert_eq!(old.layer.filename(), new.layer.filename(), "{}", context);
+                assert_eq!(old.lsn_floor, new.lsn_floor, "{}", context);
+            }
+        }
+        return Ok(new);
+    }
+
+    // HACK just testing correctness
+    fn search_new(&self, key: Key, end_lsn: Lsn) -> Result<Option<SearchResult>> {
+        // TODO I'm making two separate queries, which is 2x the cost, but that
+        //      can be avoided in varous ways. Caching latest_image queries is
+        //      probably the simplest, but combining the two data structures
+        //      might be better.
+        let latest_layer = self.index.query(key.to_i128(), end_lsn.0 - 1);
+        let latest_image = self.images.query(key.to_i128(), end_lsn.0 - 1);
+
+        // Check for exact match
+        let latest_image = if let Some(image) = latest_image {
+            let img_lsn = image.get_lsn_range().start;
+            if Lsn(img_lsn.0 + 1) == end_lsn {
+                return Ok(Some(SearchResult {
+                    layer: image,
+                    lsn_floor: img_lsn,
+                }));
+            }
+
+            // HACK just to give back ownership of latest_image to parent scope.
+            //      There's definitely a cleaner way to do it.
+            Some(image)
+        } else {
+            None
+        };
+
+        return Ok(latest_layer.map(|layer| {
+            // Compute lsn_floor
+            let mut lsn_floor = layer.get_lsn_range().start;
+            if let Some(image) = latest_image {
+                if layer.is_incremental() {
+                    lsn_floor = std::cmp::max(lsn_floor, image.get_lsn_range().start + 1)
+                }
+            }
+            SearchResult {
+                layer,
+                lsn_floor,
+            }
+        }));
+    }
+
+    // HACK just testing correctness
+    fn search_old(&self, key: Key, end_lsn: Lsn) -> Result<Option<SearchResult>> {
         // linear search
         // Find the latest image layer that covers the given key
         let mut latest_img: Option<Arc<dyn Layer>> = None;
@@ -345,19 +411,50 @@ impl LayerMap {
     /// Insert an on-disk layer
     ///
     pub fn insert_historic(&mut self, layer: Arc<dyn Layer>) {
+        let kr = layer.get_key_range();
+        let lr = layer.get_lsn_range();
+        self.index.insert(
+            kr.start.to_i128()..kr.end.to_i128(),
+            lr.start.0..lr.end.0,
+            Arc::clone(&layer),
+        );
+        if !layer.is_incremental() {
+            self.images.insert(
+                kr.start.to_i128()..kr.end.to_i128(),
+                lr.start.0..lr.end.0,
+                Arc::clone(&layer),
+            );
+        }
+
         if layer.get_key_range() == (Key::MIN..Key::MAX) {
             self.l0_delta_layers.push(layer.clone());
         }
+        // TODO remove this so insert isn't slow. I need it for now for iter_historic()
         self.historic_layers.insert(LayerRTreeObject::new(layer));
         NUM_ONDISK_LAYERS.inc();
     }
 
+    /// Must be called after a batch of insert_historic calls, before querying
+    pub fn rebuild_index(&mut self) {
+        self.index.rebuild();
+        self.images.rebuild();
+    }
+
     ///
     /// Remove an on-disk layer from the map.
     ///
     /// This should be called when the corresponding file on disk has been deleted.
     ///
     pub fn remove_historic(&mut self, layer: Arc<dyn Layer>) {
+        let kr = layer.get_key_range();
+        let lr = layer.get_lsn_range();
+        self.index
+            .remove(kr.start.to_i128()..kr.end.to_i128(), lr.start.0..lr.end.0);
+        if !layer.is_incremental() {
+            self.images
+                .remove(kr.start.to_i128()..kr.end.to_i128(), lr.start.0..lr.end.0);
+        }
+
         if layer.get_key_range() == (Key::MIN..Key::MAX) {
             let len_before = self.l0_delta_layers.len();
 
@@ -586,3 +683,5 @@ impl LayerMap {
         Ok(())
     }
 }
+
+// TODO add layer map tests

pageserver/src/tenant/timeline.rs

@@ -931,6 +931,7 @@ impl Timeline {
                 trace!("found layer {}", layer.filename().display());
                 total_physical_size += layer.path().metadata()?.len();
                 layers.insert_historic(Arc::new(layer));
+                layers.rebuild_index();
                 num_layers += 1;
             } else if let Some(deltafilename) = DeltaFileName::parse_str(&fname) {
                 // Create a DeltaLayer struct for each delta file.
@@ -955,6 +956,7 @@ impl Timeline {
                 trace!("found layer {}", layer.filename().display());
                 total_physical_size += layer.path().metadata()?.len();
                 layers.insert_historic(Arc::new(layer));
+                layers.rebuild_index();
                 num_layers += 1;
             } else if fname == METADATA_FILE_NAME || fname.ends_with(".old") {
                 // ignore these
@@ -1085,10 +1087,11 @@ impl Timeline {
                 let image_layer =
                     ImageLayer::new(self.conf, self.timeline_id, self.tenant_id, &imgfilename);
 
-                self.layers
-                    .write()
-                    .unwrap()
-                    .insert_historic(Arc::new(image_layer));
+                {
+                    let mut layers = self.layers.write().unwrap();
+                    layers.insert_historic(Arc::new(image_layer));
+                    layers.rebuild_index();
+                }
                 self.metrics.current_physical_size_gauge.add(sz);
             } else if let Some(deltafilename) = DeltaFileName::parse_str(fname) {
                 // Create a DeltaLayer struct for each delta file.
@@ -1115,10 +1118,11 @@ impl Timeline {
                 let delta_layer =
                     DeltaLayer::new(self.conf, self.timeline_id, self.tenant_id, &deltafilename);
 
-                self.layers
-                    .write()
-                    .unwrap()
-                    .insert_historic(Arc::new(delta_layer));
+                {
+                    let mut layers = self.layers.write().unwrap();
+                    layers.insert_historic(Arc::new(delta_layer));
+                    layers.rebuild_index();
+                }
                 self.metrics.current_physical_size_gauge.add(sz);
             } else {
                 bail!("unexpected layer filename in remote storage: {}", fname);
@@ -1811,6 +1815,7 @@ impl Timeline {
         {
             let mut layers = self.layers.write().unwrap();
             layers.insert_historic(Arc::new(new_delta));
+            layers.rebuild_index();
         }
 
         // update the timeline's physical size
@@ -1970,6 +1975,7 @@ impl Timeline {
             self.metrics.current_physical_size_gauge.add(metadata.len());
             layers.insert_historic(Arc::new(l));
         }
+        layers.rebuild_index();
         drop(layers);
         timer.stop_and_record();
 
@@ -2276,6 +2282,7 @@ impl Timeline {
 
             new_layer_paths.insert(new_delta_path, LayerFileMetadata::new(metadata.len()));
             layers.insert_historic(Arc::new(l));
+            layers.rebuild_index();
         }
 
         // Now that we have reshuffled the data to set of new delta layers, we can
@@ -2291,6 +2298,7 @@ impl Timeline {
             l.delete()?;
             layers.remove_historic(l);
         }
+        layers.rebuild_index();
         drop(layers);
 
         // Also schedule the deletions in remote storage
@@ -2587,6 +2595,7 @@ impl Timeline {
             layers.remove_historic(doomed_layer);
             result.layers_removed += 1;
         }
+        layers.rebuild_index();
 
         info!(
             "GC completed removing {} layers, cutoff {}",

pgxn/neon/pagestore_smgr.c

@@ -1780,6 +1780,17 @@ neon_read_at_lsn(RelFileNode rnode, ForkNumber forkNum, BlockNumber blkno,
 												  &request_lsn);
 			slot = GetPrfSlot(ring_index);
 		}
+		else
+		{
+			/*
+			 * Empty our reference to the prefetch buffer's hash entry.
+			 * When we wait for prefetches, the entry reference is invalidated by 
+			 * potential updates to the hash, and when we reconnect to the 
+			 * pageserver the prefetch we're waiting for may be dropped,
+			 * in which case we need to retry and take the branch above.
+			 */
+			entry = NULL;
+		}
 
 		Assert(slot->my_ring_index == ring_index);
 		Assert(MyPState->ring_last <= ring_index &&

test_runner/regress/test_import.py

@@ -120,6 +120,12 @@ def test_import_from_vanilla(test_output_dir, pg_bin, vanilla_pg, neon_env_build
             ]
         )
 
+    # Importing empty file fails
+    empty_file = os.path.join(test_output_dir, "empty_file")
+    with open(empty_file, "w") as _:
+        with pytest.raises(Exception):
+            import_tar(empty_file, empty_file)
+
     # Importing corrupt backup fails
     with pytest.raises(Exception):
         import_tar(corrupt_base_tar, wal_tar)

workspace_hack/Cargo.toml

@@ -16,7 +16,6 @@ publish = false
 ahash = { version = "0.7", features = ["std"] }
 anyhow = { version = "1", features = ["backtrace", "std"] }
 bytes = { version = "1", features = ["serde", "std"] }
-chrono = { version = "0.4", features = ["clock", "iana-time-zone", "js-sys", "oldtime", "serde", "std", "time", "wasm-bindgen", "wasmbind", "winapi"] }
 clap = { version = "4", features = ["color", "derive", "error-context", "help", "std", "string", "suggestions", "usage"] }
 crossbeam-utils = { version = "0.8", features = ["once_cell", "std"] }
 either = { version = "1", features = ["use_std"] }
@@ -31,7 +30,7 @@ log = { version = "0.4", default-features = false, features = ["serde", "std"] }
 memchr = { version = "2", features = ["std"] }
 nom = { version = "7", features = ["alloc", "std"] }
 num-bigint = { version = "0.4", features = ["std"] }
-num-integer = { version = "0.1", default-features = false, features = ["i128", "std"] }
+num-integer = { version = "0.1", features = ["i128", "std"] }
 num-traits = { version = "0.2", features = ["i128", "libm", "std"] }
 prost-93f6ce9d446188ac = { package = "prost", version = "0.10", features = ["prost-derive", "std"] }
 prost-a6292c17cd707f01 = { package = "prost", version = "0.11", features = ["prost-derive", "std"] }
@@ -45,6 +44,7 @@ stable_deref_trait = { version = "1", features = ["alloc", "std"] }
 time = { version = "0.3", features = ["alloc", "formatting", "itoa", "macros", "parsing", "std", "time-macros"] }
 tokio = { version = "1", features = ["bytes", "fs", "io-std", "io-util", "libc", "macros", "memchr", "mio", "net", "num_cpus", "once_cell", "process", "rt", "rt-multi-thread", "signal-hook-registry", "socket2", "sync", "time", "tokio-macros"] }
 tokio-util = { version = "0.7", features = ["codec", "io", "io-util", "tracing"] }
+tower = { version = "0.4", features = ["__common", "balance", "buffer", "discover", "futures-core", "futures-util", "indexmap", "limit", "load", "log", "make", "pin-project", "pin-project-lite", "rand", "ready-cache", "retry", "slab", "timeout", "tokio", "tokio-util", "tracing", "util"] }
 tracing = { version = "0.1", features = ["attributes", "log", "std", "tracing-attributes"] }
 tracing-core = { version = "0.1", features = ["once_cell", "std"] }