feat: add payload offloading to GCS read/write paths

via a new payload_link column (currently Spanner only)

Closes STOR-578
Closes STOR-579

Author: pjenvey

Cargo.lock

@@ -39,8 +39,11 @@ actix-http = "3"
 actix-rt = "2"
 actix-cors = "0.7"
 glean = { path = "../glean" }
+google-cloud-auth = "1.13"
+google-cloud-storage = "1.15"
 hawk = "5.0"
 mime = "0.3"
+wkt = { package = "google-cloud-wkt", version = "1.5" }
 # pin to 0.19: https://github.com/getsentry/sentry-rust/issues/277
 syncserver-common = { path = "../syncserver-common" }
 syncserver-db-common = { path = "../syncserver-db-common" }
@@ -60,6 +63,7 @@ validator_derive = "0.20"
 woothee = "0.13"
 
 [dev-dependencies]
+httptest = "0.16"
 temp-env.workspace = true
 tokenserver-auth = { path = "../tokenserver-auth", features = ["test-support"] }
 

syncserver/Cargo.toml

@@ -69,6 +69,9 @@ pub enum ApiErrorKind {
     #[error("{}", _0)]
     Db(DbError),
 
+    #[error("{}", _0)]
+    GCS(#[from] google_cloud_storage::Error),
+
     #[error("HAWK authentication error: {}", _0)]
     Hawk(HawkError),
 
@@ -172,7 +175,7 @@ impl From<ApiErrorKind> for ApiError {
         let status = match &kind {
             ApiErrorKind::Db(error) => error.status,
             ApiErrorKind::Hawk(_) => StatusCode::UNAUTHORIZED,
-            ApiErrorKind::NoServerState | ApiErrorKind::Internal(_) => {
+            ApiErrorKind::NoServerState | ApiErrorKind::Internal(_) | ApiErrorKind::GCS(_) => {
                 StatusCode::INTERNAL_SERVER_ERROR
             }
             ApiErrorKind::Validation(error) => error.status,
@@ -234,6 +237,7 @@ impl Serialize for ApiErrorKind {
     {
         match *self {
             ApiErrorKind::Db(ref error) => serialize_string_to_array(serializer, error),
+            ApiErrorKind::GCS(ref error) => serialize_string_to_array(serializer, error),
             ApiErrorKind::Hawk(ref error) => serialize_string_to_array(serializer, error),
             ApiErrorKind::Internal(ref description) => {
                 serialize_string_to_array(serializer, description)
@@ -268,6 +272,7 @@ impl From<DbError> for ApiError {
     }
 }
 
+from_error!(google_cloud_storage::Error, ApiError, ApiErrorKind::GCS);
 from_error!(HawkError, ApiError, ApiErrorKind::Hawk);
 from_error!(ValidationError, ApiError, ApiErrorKind::Validation);
 

syncserver/src/error.rs

@@ -67,6 +67,18 @@ pub struct ServerState {
     pub glean_logger: Arc<GleanEventsLogger>,
 
     pub glean_enabled: bool,
+
+    /// Optional Google Cloud Storage bucket for off-loading BSO payloads.
+    pub gcs_payload_bucket: Option<String>,
+
+    /// Collections whose BSO payloads are off-loaded to GCS.
+    pub gcs_payload_offload_collections: Arc<Vec<String>>,
+
+    /// Override the GCS endpoint URL (for testing). When set the GCS client
+    /// is built with `.with_endpoint(...)` + anonymous credentials.
+    /// Debug-builds only; not available in release.
+    #[cfg(debug_assertions)]
+    pub gcs_endpoint: Option<String>,
 }
 
 pub fn cfg_path(path: &str) -> String {
@@ -379,6 +391,11 @@ impl Server {
             app_channel: settings.environment.clone(),
         });
         let glean_enabled = settings.syncstorage.glean_enabled;
+        let gcs_payload_bucket = settings.syncstorage.gcs_payload_bucket.clone();
+        let gcs_payload_offload_collections =
+            Arc::new(settings.syncstorage.gcs_payload_offload_collections.clone());
+        #[cfg(debug_assertions)]
+        let gcs_endpoint = settings.syncstorage.gcs_endpoint.clone();
         let worker_thread_count =
             calculate_worker_max_blocking_threads(settings.worker_max_blocking_threads);
         let limits = Arc::new(settings.syncstorage.limits);
@@ -426,6 +443,10 @@ impl Server {
                 deadman: Arc::clone(&deadman),
                 glean_logger: Arc::clone(&glean_logger),
                 glean_enabled,
+                gcs_payload_bucket: gcs_payload_bucket.clone(),
+                gcs_payload_offload_collections: Arc::clone(&gcs_payload_offload_collections),
+                #[cfg(debug_assertions)]
+                gcs_endpoint: gcs_endpoint.clone(),
             };
 
             build_app!(

syncserver/src/server/mod.rs

@@ -14,6 +14,11 @@ use chrono::offset::Utc;
 use hawk::{self, Credentials, Key, RequestBuilder};
 use hmac::{Hmac, KeyInit, Mac};
 use http::StatusCode;
+use httptest::{
+    Expectation, Server,
+    matchers::{all_of, contains, matches, request, url_decoded},
+    responders::status_code,
+};
 use lazy_static::lazy_static;
 use serde::de::DeserializeOwned;
 use serde_json::{Value, json};
@@ -113,6 +118,11 @@ async fn get_test_state(settings: &Settings) -> ServerState {
         deadman: Arc::new(RwLock::new(Deadman::from(&settings.syncstorage))),
         glean_logger,
         glean_enabled: settings.syncstorage.glean_enabled,
+        gcs_payload_bucket: settings.syncstorage.gcs_payload_bucket.clone(),
+        gcs_payload_offload_collections: Arc::new(
+            settings.syncstorage.gcs_payload_offload_collections.clone(),
+        ),
+        gcs_endpoint: settings.syncstorage.gcs_endpoint.clone(),
     }
 }
 
@@ -446,6 +456,7 @@ async fn post_collection() {
         id: "foo".to_string(),
         sortindex: Some(0),
         payload: Some("bar".to_string()),
+        payload_link: None,
         ttl: Some(31_536_000),
     }]);
     let bytes =
@@ -892,3 +903,79 @@ async fn error_endpoint_logging_check() {
     )
     .await;
 }
+
+#[actix_rt::test]
+async fn put_bso_offloads_to_gcs() {
+    let mut settings = get_test_settings();
+    if !settings.syncstorage.uses_spanner() {
+        // TODO: should be Spanner only (for now)
+        return;
+    }
+
+    // Mock GCS: match the JSON multipart upload to bsos's bucket. The matcher
+    // inspects the multipart body for `committed=false` metadata and a
+    // `customTime` field; if either is missing the expectation goes
+    // unsatisfied and Server::Drop panics.
+    let server = Server::run();
+    server.expect(
+        Expectation::matching(all_of![
+            request::method_path("POST", "/upload/storage/v1/b/test-bucket/o"),
+            request::query(url_decoded(contains(("uploadType", "multipart")))),
+            request::body(matches(r#""committed""#)),
+            request::body(matches(r#""false""#)),
+            request::body(matches(r#""customTime""#)),
+        ])
+        .respond_with(
+            status_code(200)
+                .append_header("content-type", "application/json")
+                .body(
+                    serde_json::json!({
+                        "name": "test-object",
+                        "bucket": "test-bucket",
+                    })
+                    .to_string(),
+                ),
+        ),
+    );
+
+    settings.syncstorage.gcs_payload_bucket = Some("test-bucket".to_owned());
+    settings.syncstorage.gcs_payload_offload_collections = vec!["bookmarks".to_owned()];
+    settings.syncstorage.gcs_endpoint = Some(format!("http://{}", server.addr()));
+
+    let app = init_app!(settings).await;
+
+    let req = create_request(
+        http::Method::PUT,
+        "/1.5/42/storage/bookmarks/wibble",
+        None,
+        Some(json!({ "payload": "hello world" })),
+    )
+    .to_request();
+    let resp = app
+        .call(req)
+        .await
+        .expect("Could not get resp in put_bso_offloads_to_gcs");
+    assert!(
+        resp.response().status().is_success(),
+        "put_bso failed: {:?}",
+        resp.response()
+    );
+
+    // mock shouldn't see tabs offloaded to GCS
+    let req = create_request(
+        http::Method::PUT,
+        "/1.5/42/storage/tabs/wobble",
+        None,
+        Some(json!({ "payload": "hello world" })),
+    )
+    .to_request();
+    let resp = app
+        .call(req)
+        .await
+        .expect("Could not get resp in put_bso_offloads_to_gcs");
+    assert!(
+        resp.response().status().is_success(),
+        "put_bso failed: {:?}",
+        resp.response()
+    );
+}

syncserver/src/server/test.rs

@@ -15,6 +15,10 @@ pub struct BatchBsoBody {
     pub payload: Option<String>,
     #[validate(custom(function = "validate_body_bso_ttl"))]
     pub ttl: Option<u32>,
+    /// Populated server-side after a successful GCS payload upload; never
+    /// accepted from the client.
+    #[serde(default, skip_deserializing)]
+    pub payload_link: Option<String>,
 }
 
 impl BatchBsoBody {
@@ -50,6 +54,7 @@ impl From<BatchBsoBody> for PostCollectionBso {
             id: b.id,
             sortindex: b.sortindex,
             payload: b.payload,
+            payload_link: b.payload_link,
             ttl: b.ttl,
         }
     }

syncserver/src/web/extractors/batch_bso_body.rs

@@ -68,6 +68,10 @@ pub fn make_state() -> ServerState {
         deadman: Arc::new(RwLock::new(Deadman::default())),
         glean_logger,
         glean_enabled: syncstorage_settings.glean_enabled,
+        gcs_payload_bucket: None,
+        gcs_payload_offload_collections: Arc::new(Vec::new()),
+        #[cfg(debug_assertions)]
+        gcs_endpoint: None,
     }
 }