bug: write bsos contained within a commit after the batch has been commited.

Cargo.toml

@@ -49,7 +49,7 @@ hostname = "0.3.1"
 hkdf = "0.10"
 hmac = "0.10"
 jsonwebtoken = "7.2.0"
-log = { version = "0.4", features = ["max_level_info", "release_max_level_info"] }
+log = { version = "0.4", features = ["max_level_debug", "release_max_level_info"] }
 mime = "0.3"
 num_cpus = "1"
 # must match what's used by googleapis-raw

src/web/handlers.rs

@@ -320,77 +320,63 @@ pub async fn post_collection_batch(
         .await?
     };
 
-    let commit = breq.commit;
     let user_id = coll.user_id.clone();
     let collection = coll.collection.clone();
 
     let mut success = vec![];
     let mut failed = coll.bsos.invalid;
     let bso_ids: Vec<_> = coll.bsos.valid.iter().map(|bso| bso.id.clone()).collect();
 
-    let result = if commit && !coll.bsos.valid.is_empty() {
-        // There's pending items to append to the batch but since we're
-        // committing, write them to bsos immediately. Otherwise under
-        // Spanner we would pay twice the mutations for those pending
-        // items (once writing them to to batch_bsos, then again
-        // writing them to bsos)
-        trace!("Batch: Committing {}", &new_batch.id);
-        db.post_bsos(params::PostBsos {
-            user_id: coll.user_id.clone(),
-            collection: coll.collection.clone(),
-            // XXX: why does BatchBsoBody exist (it's the same struct
-            // as PostCollectionBso)?
-            bsos: coll
-                .bsos
-                .valid
-                .into_iter()
-                .map(|batch_bso| params::PostCollectionBso {
-                    id: batch_bso.id,
-                    sortindex: batch_bso.sortindex,
-                    payload: batch_bso.payload,
-                    ttl: batch_bso.ttl,
-                })
-                .collect(),
-            for_batch: true,
-            failed: Default::default(),
-        })
-        .await
-        .map(|_| ())
-    } else {
-        // We're not yet to commit the accumulated batch, but there are some
-        // additional records we need to add.
-        trace!("Batch: Appending to {}", &new_batch.id);
-        db.append_to_batch(params::AppendToBatch {
-            user_id: coll.user_id.clone(),
-            collection: coll.collection.clone(),
-            batch: new_batch.clone(),
-            bsos: coll.bsos.valid.into_iter().map(From::from).collect(),
-        })
-        .await
-    };
-
-    // collect up the successful and failed bso_ids into a response.
-    match result {
-        Ok(_) => success.extend(bso_ids),
-        Err(e) if e.is_conflict() => return Err(e.into()),
-        Err(apperr) => {
-            if let ApiErrorKind::Db(dberr) = apperr.kind() {
-                // If we're over quota, return immediately with a 403 to let the client know.
-                // Otherwise the client will simply keep retrying records.
-                if let DbErrorKind::Quota = dberr.kind() {
-                    return Err(apperr.into());
+    let mut resp: Value = json!({});
+
+    macro_rules! handle_result {
+            // collect up the successful and failed bso_ids into a response.
+            ( $r: expr) => {
+            match $r {
+                Ok(_) => success.extend(bso_ids.clone()),
+                Err(e) if e.is_conflict() => return Err(e.into()),
+                Err(apperr) => {
+                    if let ApiErrorKind::Db(dberr) = apperr.kind() {
+                        // If we're over quota, return immediately with a 403 to let the client know.
+                        // Otherwise the client will simply keep retrying records.
+                        if let DbErrorKind::Quota = dberr.kind() {
+                            return Err(apperr.into());
+                        }
+                    };
+                    failed.extend(
+                        bso_ids
+                            .clone()
+                            .into_iter()
+                            .map(|id| (id, "db error".to_owned())),
+                    )
                 }
             };
-            failed.extend(bso_ids.into_iter().map(|id| (id, "db error".to_owned())))
         }
-    };
-
-    let mut resp = json!({
-        "success": success,
-        "failed": failed,
-    });
+    }
 
+    // If we're not committing the current set of records yet.
     if !breq.commit {
+        // and there are bsos included in this message.
+        if !coll.bsos.valid.is_empty() {
+            // Append the data to the requested batch.
+            let result = {
+                dbg!("Batch: Appending to {}", &new_batch.id);
+                db.append_to_batch(params::AppendToBatch {
+                    user_id: coll.user_id.clone(),
+                    collection: coll.collection.clone(),
+                    batch: new_batch.clone(),
+                    bsos: coll.bsos.valid.into_iter().map(From::from).collect(),
+                })
+                .await
+            };
+            handle_result!(result);
+        }
+
+        // Return the batch append response without committing the current
+        // batch to the BSO table.
+        resp["success"] = json!(success);
+        resp["failed"] = json!(failed);
+
         resp["batch"] = json!(&new_batch.id);
         return Ok(HttpResponse::Accepted().json(resp));
     }
@@ -406,6 +392,8 @@ pub async fn post_collection_batch(
 
     // TODO: validate *actual* sizes of the batch items
     // (max_total_records, max_total_bytes)
+    //
+    // First, write the pending batch BSO data into the BSO table.
     let modified = if let Some(batch) = batch {
         db.commit_batch(params::CommitBatch {
             user_id: user_id.clone(),
@@ -418,6 +406,41 @@ pub async fn post_collection_batch(
         return Err(ApiError::from(err).into());
     };
 
+    // Then, write the BSOs contained in the commit request into the BSO table.
+    // (This presumes that the BSOs contained in the final "commit" message are
+    // newer, and thus more "correct", than any prior BSO info that may have been
+    // included in the prior batch creation messages. The client shouldn't really
+    // be including BSOs with the commit message, however it does and we should
+    // handle that case.)
+    if !coll.bsos.valid.is_empty() {
+        trace!("Batch: writing commit message bsos");
+        let result = db
+            .post_bsos(params::PostBsos {
+                user_id: coll.user_id.clone(),
+                collection: coll.collection.clone(),
+                bsos: coll
+                    .bsos
+                    .valid
+                    .into_iter()
+                    .map(|batch_bso| params::PostCollectionBso {
+                        id: batch_bso.id,
+                        sortindex: batch_bso.sortindex,
+                        payload: batch_bso.payload,
+                        ttl: batch_bso.ttl,
+                    })
+                    .collect(),
+                for_batch: false,
+                failed: Default::default(),
+            })
+            .await
+            .map(|_| ());
+
+        handle_result!(result);
+
+        resp["success"] = json!(success);
+        resp["failed"] = json!(failed);
+    }
+
     resp["modified"] = json!(modified);
     trace!("Batch: Returning result: {}", &resp);
     Ok(HttpResponse::build(StatusCode::OK)

tools/integration_tests/test_storage.py

@@ -1582,6 +1582,35 @@ def test_batches(self):
         resp3 = self.app.get(endpoint + '/e')
         self.assertEquals(committed, resp3.json['modified'])
 
+
+    def test_aaa_batch_commit_collision(self):
+        # It's possible that a batch contain a BSO inside a batch as well
+        # as inside the final "commit" message. This is a bit of a problem
+        # for spanner because of conflicting ways that the data is written
+        # to the database and the discoverability of IDs in previously
+        # submitted batches.
+        endpoint = self.root + '/storage/xxx_col2'
+        orig = "Letting the days go by"
+        repl = "Same as it ever was"
+
+        batch_num = self.retry_post_json(
+            endpoint + "?batch=true",
+            [{"id":"b0", "payload": orig}]
+        ).json["batch"]
+
+        resp = self.retry_post_json(
+            endpoint + "?batch={}&commit=true".format(batch_num),
+            [{"id":"b0", "payload": repl}]
+        )
+
+        # this should succeed, using the newerer payload value.
+        assert resp.json["failed"] == {}, "batch commit failed"
+        assert resp.json["success"] == ["b0"], "batch commit id incorrect"
+        resp = self.app.get(endpoint+"?full=1")
+        assert resp.json[0].get(
+            "payload") == repl, "wrong payload returned"
+
+
     def test_we_dont_need_no_stinkin_batches(self):
         endpoint = self.root + '/storage/xxx_col2'