Add "tumble" distributed transaction processing service for appengine.

appengine/tumble/tumble.go

Index: appengine/tumble/tumble.go
diff --git a/appengine/tumble/tumble.go b/appengine/tumble/tumble.go
new file mode 100644
index 0000000000000000000000000000000000000000..6b95190c04c526f54674a3152f79df4f72d3c225
--- /dev/null
+++ b/appengine/tumble/tumble.go
@@ -0,0 +1,61 @@
+// Copyright 2015 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+package tumble
+
+import (
+	"fmt"
+
+	"github.com/luci/gae/filter/txnBuf"
+	"github.com/luci/gae/service/datastore"
+	"golang.org/x/net/context"
+)
+
+// EnterTransaction is the method to use when doing transactional operations
+// in the tumble ecosystem. It allows work to be done in the entity group
+// specified by `fromRoot`, and any returned Mutation objects will be
+// transactionally queued for the tumble backend.
+//
+// Usually this is called from your application's handlers to begin a tumble
+// state machine as a result of some API interaction.
+func EnterTransaction(c context.Context, fromRoot *datastore.Key, f func(context.Context) ([]Mutation, error)) error {
+	shardSet, _, err := enterTransactionInternal(txnBuf.FilterRDS(c), fromRoot, f)
+	if err != nil {
+		return err
+	}
+	fireTasks(c, shardSet)
+	return nil
+}
+
+func enterTransactionInternal(c context.Context, fromRoot *datastore.Key, f func(context.Context) ([]Mutation, error)) (map[uint64]struct{}, uint, error) {

[iannucci, 2015/10/10 17:51:53]

split this up so that when we're in the process loop we can commit the new
mutations along with the originating mutation (in an inner transaction). This
should prevent the transaction from becoming too large and then failing to
commit the new mutations, causing the whole batch to fail. The original
buffering scheme was devised when we didn't have the ability to do nested
transactions.

+	if fromRoot == nil {
+		return nil, 0, fmt.Errorf("tumble: Passing nil as fromRoot is illegal")
+	}
+
+	shardSet := map[uint64]struct{}(nil)
+	numNewMuts := uint(0)
+
+	err := datastore.Get(c).RunInTransaction(func(c context.Context) error {
+		// do a Get on the fromRoot to ensure that this transaction is associated
+		// with that entity group.
+		_ = datastore.Get(c).Get(datastore.PropertyMap{
+			"$key": {datastore.MkPropertyNI(fromRoot)},

[Vadim Sh., 2015/10/12 20:05:14]

is it how hypothetical Exists(key) looks?

[iannucci, 2015/10/13 02:39:46]

Yeah, it's kinda ugly. Do you think it's worth adding an Exists to
datastore.Interface?

[Vadim Sh., 2015/10/13 02:46:20]

Yes. 

We actually needed this in isolate server (here
https://github.com/luci/luci-py/blob/master/appengine/isolate/handlers_endpoi...)
and weren't smart enough to implement over what Python SDK provides.

Or I misunderstood and it still fetches everything, just doesn't deserialize it?
Is there a way to fetch key only?

+		})
+
+		muts, err := f(c)
+		if err != nil {
+			return err
+		}
+
+		numNewMuts = uint(len(muts))
+		shardSet, err = putMutations(c, fromRoot, muts)
+
+		return err
+	}, nil)
+	if err != nil {
+		return nil, 0, err
+	}
+
+	return shardSet, numNewMuts, nil
+}