refactor(db): use Spanner INSERT OR UPDATE DML

syncserver-settings/src/lib.rs

@@ -170,7 +170,6 @@ impl Settings {
         settings.port = 8000;
         settings.syncstorage.database_pool_max_size = 1;
         settings.syncstorage.database_use_test_transactions = true;
-        settings.syncstorage.database_spanner_use_mutations = false;
         settings.syncstorage.database_pool_connection_max_idle = Some(300);
         settings.syncstorage.database_pool_connection_lifespan = Some(300);
         settings

syncstorage-db/src/tests/batch.rs

@@ -375,6 +375,128 @@ async fn test_append_async_w_empty_string() -> Result<(), DbError> {
     .await
 }
 
+#[tokio::test]
+async fn test_append_upsert_overwrites_same_batch_bso_id() -> Result<(), DbError> {
+    let settings = Settings::test_settings().syncstorage;
+    with_test_transaction(settings, async |db: &mut dyn Db<Error = DbError>| {
+        let uid = 1;
+        let coll = "clients";
+        let bid = "b0";
+        let first_payload = "wibble";
+        let second_payload = "over 9000";
+
+        let new_batch = db.create_batch(cb(uid, coll, vec![])).await?;
+        db.append_to_batch(ab(
+            uid,
+            coll,
+            new_batch.clone(),
+            vec![postbso(bid, Some(first_payload), Some(10), None)],
+        ))
+        .await?;
+        db.append_to_batch(ab(
+            uid,
+            coll,
+            new_batch.clone(),
+            vec![postbso(bid, Some(second_payload), None, None)],
+        ))
+        .await?;
+
+        let batch = db
+            .get_batch(gb(uid, coll, new_batch.id.clone()))
+            .await?
+            .unwrap();
+        db.commit_batch(params::CommitBatch {
+            user_id: hid(uid),
+            collection: coll.to_owned(),
+            batch,
+        })
+        .await?;
+
+        let bso = db.get_bso(gbso(uid, coll, bid)).await?.unwrap();
+        assert_eq!(bso.payload, second_payload);
+        assert_eq!(bso.sortindex, Some(10));
+        Ok(())
+    })
+    .await
+}
+
+#[tokio::test]
+async fn test_commit_batch_partial_overlap() -> Result<(), DbError> {
+    let settings = Settings::test_settings().syncstorage;
+    with_test_transaction(settings, async |db: &mut dyn Db<Error = DbError>| {
+        let uid = 1;
+        let coll = "clients";
+        let bid_overlap = "b_overlap";
+        let bid_existing_only = "b_existing";
+        let bid_new = "b_new";
+
+        let original_overlap_payload = "fizz";
+        let existing_only_payload = "buzz";
+        let updated_overlap_payload = "quux";
+        let new_payload = "new hotness";
+
+        db.put_bso(pbso(
+            uid,
+            coll,
+            bid_overlap,
+            Some(original_overlap_payload),
+            Some(1),
+            None,
+        ))
+        .await?;
+        db.put_bso(pbso(
+            uid,
+            coll,
+            bid_existing_only,
+            Some(existing_only_payload),
+            Some(2),
+            None,
+        ))
+        .await?;
+
+        let new_batch = db.create_batch(cb(uid, coll, vec![])).await?;
+        db.append_to_batch(ab(
+            uid,
+            coll,
+            new_batch.clone(),
+            vec![
+                postbso(bid_overlap, Some(updated_overlap_payload), None, None),
+                postbso(bid_new, Some(new_payload), Some(3), None),
+            ],
+        ))
+        .await?;
+
+        let batch = db
+            .get_batch(gb(uid, coll, new_batch.id.clone()))
+            .await?
+            .unwrap();
+        db.commit_batch(params::CommitBatch {
+            user_id: hid(uid),
+            collection: coll.to_owned(),
+            batch,
+        })
+        .await?;
+
+        let overlap = db.get_bso(gbso(uid, coll, bid_overlap)).await?.unwrap();
+        assert_eq!(overlap.payload, updated_overlap_payload);
+        assert_eq!(overlap.sortindex, Some(1));
+
+        let existing = db
+            .get_bso(gbso(uid, coll, bid_existing_only))
+            .await?
+            .unwrap();
+        assert_eq!(existing.payload, existing_only_payload);
+        assert_eq!(existing.sortindex, Some(2));
+
+        let new = db.get_bso(gbso(uid, coll, bid_new)).await?.unwrap();
+        assert_eq!(new.payload, new_payload);
+        assert_eq!(new.sortindex, Some(3));
+
+        Ok(())
+    })
+    .await
+}
+
 #[tokio::test]
 async fn pretouch() -> Result<(), DbError> {
     let settings = Settings::test_settings().syncstorage;

syncstorage-db/src/tests/db.rs

@@ -1104,3 +1104,46 @@ async fn heartbeat() -> Result<(), DbError> {
     })
     .await
 }
+
+#[tokio::test]
+async fn put_bso_preserves_payload_on_sortindex_only_update() -> Result<(), DbError> {
+    with_test_transaction(None, async |db: &mut dyn Db<Error = DbError>| {
+        let uid = *UID;
+        let coll = "clients";
+        let bid = "preserve_payload";
+
+        db.put_bso(pbso(uid, coll, bid, Some("quux"), Some(10), None))
+            .await?;
+
+        db.put_bso(pbso(uid, coll, bid, None, Some(20), None))
+            .await?;
+
+        let bso = db.get_bso(gbso(uid, coll, bid)).await?.unwrap();
+        assert_eq!(bso.payload, "quux");
+        assert_eq!(bso.sortindex, Some(20));
+        Ok(())
+    })
+    .await
+}
+
+#[tokio::test]
+async fn put_bso_preserves_expiry_on_partial_update() -> Result<(), DbError> {
+    with_test_transaction(None, async |db: &mut dyn Db<Error = DbError>| {
+        let uid = *UID;
+        let coll = "clients";
+        let bid = "preserve_expiry";
+
+        db.put_bso(pbso(uid, coll, bid, Some("quux"), None, Some(3600)))
+            .await?;
+        let original_expiry = db.get_bso(gbso(uid, coll, bid)).await?.unwrap().expiry;
+
+        db.put_bso(pbso(uid, coll, bid, Some("baz"), None, None))
+            .await?;
+
+        let bso = db.get_bso(gbso(uid, coll, bid)).await?.unwrap();
+        assert_eq!(bso.payload, "baz");
+        assert_eq!(bso.expiry, original_expiry);
+        Ok(())
+    })
+    .await
+}

syncstorage-settings/src/lib.rs

@@ -81,8 +81,6 @@ pub struct Settings {
     pub database_pool_sweeper_task_interval: u32,
     #[cfg(debug_assertions)]
     pub database_use_test_transactions: bool,
-    #[cfg(debug_assertions)]
-    pub database_spanner_use_mutations: bool,
     /// Whether leader aware router headers are sent to Spanner
     pub database_spanner_route_to_leader: bool,
 
@@ -119,8 +117,6 @@ impl Default for Settings {
             database_pool_connection_timeout: Some(30),
             #[cfg(debug_assertions)]
             database_use_test_transactions: false,
-            #[cfg(debug_assertions)]
-            database_spanner_use_mutations: true,
             database_spanner_route_to_leader: false,
             limits: ServerLimits::default(),
             statsd_label: "syncstorage".to_string(),

syncstorage-spanner/src/db/batch_commit_upsert.sql

@@ -0,0 +1,37 @@
+-- bsos.payload/modified/expiry are NOT NULL with no schema default, so the
+-- insert of INSERT OR UPDATE must supply a value for every column.  Each
+-- driving `batch_bsos` row LEFT JOINs to a subquery `existing` row.  On update
+-- it provides values to COALESCE over; on insert the join misses, `existing.*`
+-- is NULL, and the COALESCE fallback applies.
+INSERT OR UPDATE INTO bsos
+    (fxa_uid, fxa_kid, collection_id, bso_id, sortindex, payload, modified, expiry)
+SELECT
+    bb.fxa_uid,
+    bb.fxa_kid,
+    bb.collection_id,
+    bb.batch_bso_id,
+    COALESCE(bb.sortindex, existing.sortindex),
+    COALESCE(bb.payload, existing.payload, ''),
+    @timestamp,
+    COALESCE(
+        TIMESTAMP_ADD(@timestamp, INTERVAL bb.ttl SECOND),
+        existing.expiry,
+        TIMESTAMP_ADD(@timestamp, INTERVAL @default_bso_ttl SECOND)
+    )
+  FROM batch_bsos AS bb
+  LEFT JOIN (
+      SELECT fxa_uid, fxa_kid, collection_id, bso_id,
+             sortindex, payload, expiry
+        FROM bsos
+       WHERE fxa_uid = @fxa_uid
+         AND fxa_kid = @fxa_kid
+         AND collection_id = @collection_id
+  ) AS existing
+    ON existing.fxa_uid = bb.fxa_uid
+   AND existing.fxa_kid = bb.fxa_kid
+   AND existing.collection_id = bb.collection_id
+   AND existing.bso_id = bb.batch_bso_id
+ WHERE bb.fxa_uid = @fxa_uid
+   AND bb.fxa_kid = @fxa_kid
+   AND bb.collection_id = @collection_id
+   AND bb.batch_id = @batch_id