GCS Provider Bytes Range Headers (#12855)

## Problem
 
Bytes range headers are not yet implemented for the GCS JSON API
interface in Neon,
[affecting](489c7a20f4/safekeeper/src/wal_backup.rs (L623))
`read_object` in SafeKeepers' `wal_backup.rs`, when reading partial
segments back from remote storage.

## Summary of changes
 * Handle bytes range header for GCS JSON API
 * Testing

libs/remote_storage/src/config.rs

@@ -339,7 +339,7 @@ timeout = '5s'";
     fn test_gcs_parsing() {
         let toml = "\
     bucket_name = 'foo-bar'
-    prefix_in_bucket = '/pageserver'
+    prefix_in_bucket = 'pageserver/'
     ";
 
         let config = parse(toml).unwrap();

libs/remote_storage/src/gcs_bucket.rs

@@ -728,7 +728,7 @@ impl GCSBucket {
                         StatusCode::NOT_FOUND => return Err(DownloadError::NotFound),
                         _ => {
                             return Err(DownloadError::Other(anyhow::anyhow!(
-                                "GCS GET resposne contained no response body"
+                                "GCS GET response contained no response body"
                             )));
                         }
                     }
@@ -751,7 +751,17 @@ impl GCSBucket {
 
         // 2. Byte Stream request
         let mut headers = header::HeaderMap::new();
-        headers.insert(header::RANGE, header::HeaderValue::from_static("bytes=0-"));
+        let bytes_range = match &request.range {
+           Some(s) => header::HeaderValue::from_str(s).unwrap(),
+           None => header::HeaderValue::from_static("bytes=0-"),
+        };
+        
+        tracing::info!(
+            "performing object download with {:?} range header",
+            bytes_range
+        );
+        
+        headers.insert(header::RANGE, bytes_range);
 
         let encoded_path: String =
             url::form_urlencoded::byte_serialize(request.key.as_bytes()).collect();

libs/remote_storage/tests/test_real_gcs.rs

@@ -75,6 +75,28 @@ impl AsyncTestContext for EnabledGCS {
     }
 }
 
+#[test_context(EnabledGCS)]
+#[tokio::test]
+async fn gcs_get_object_bytes_range_header(ctx: &mut EnabledGCS) -> anyhow::Result<()> {
+    let cancel = CancellationToken::new();
+    let path = RemotePath::new(Utf8Path::new(
+        format!("{}/000000010000028000000086", ctx.base_prefix).as_str(),
+    ))
+    .with_context(|| "RemotePath conversion")?;
+
+    let (data, len) = upload_stream("hello, world".as_bytes().into());
+
+    ctx.client.upload(data, len, &path, None, &cancel).await?;
+
+    let opts = DownloadOpts {
+        byte_start: Bound::Included(7),
+        ..Default::default()
+    };
+    let dl_object = download_to_vec(ctx.client.download(&path, &opts, &cancel).await?).await?;
+    let s = String::from_utf8(dl_object).unwrap();
+    assert_eq!(5, s.len());
+    Ok(())
+}
 #[test_context(EnabledGCS)]
 #[tokio::test]
 async fn gcs_test_suite(ctx: &mut EnabledGCS) -> anyhow::Result<()> {