Add emulation of transaction retries to memory impl of RunInTransaction.

impl/memory/context.go

 // 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 memory
 
 import (
         "errors"
         "sync"
 
@@ skipping 117 matching lines @@
 //
 // It really should have been appengine.Context.RunInTransaction(func(tc...)),
 // but because it's not, this method is on dsImpl instead to mirror the official
 // API.
 //
 // The fake implementation also differs from the real implementation because the
 // fake TaskQueue is NOT backed by the fake Datastore. This is done to make the
 // test-access API for TaskQueue better (instead of trying to reconstitute the
 // state of the task queue from a bunch of datastore accesses).
 func (d *dsImpl) RunInTransaction(f func(context.Context) error, o *ds.TransactionOptions) error {
-        curMC := cur(d.c)
+        errNope := errors.New("nope")
 
-        txnMC := curMC.mkTxn(o)
+        // Keep in separate function for defers.
+        loopBody := func(applyForReal bool) error {
+                curMC := cur(d.c)
 
-        defer func() {
+                txnMC := curMC.mkTxn(o)
+
+                defer func() {
+                        txnMC.Lock()
+                        defer txnMC.Unlock()
+
+                        txnMC.endTxn()
+                }()
+
+                if err := f(context.WithValue(d.c, memContextKey, txnMC)); err != nil {
+                        return err
+                }
+
                 txnMC.Lock()
                 defer txnMC.Unlock()
 
-                txnMC.endTxn()
-        }()
+                if !applyForReal {
+                        // Be defensive and expect that f may return ErrConcurrentTransaction. So
+                        // use some other error to signal that we skipped commit.
+                        return errNope
+                }
 
-        if err := f(context.WithValue(d.c, memContextKey, txnMC)); err != nil {
-                return err
+                if curMC.canApplyTxn(txnMC) {
+                        curMC.applyTxn(d.c, txnMC)
+                } else {
+                        return ds.ErrConcurrentTransaction
+                }
+                return nil
         }
 
-        txnMC.Lock()
-        defer txnMC.Unlock()
-
-        if curMC.canApplyTxn(txnMC) {
-                curMC.applyTxn(d.c, txnMC)
-        } else {
-                return ds.ErrConcurrentTransaction
+        // From GAE docs for TransactionOptions: "If omitted, it defaults to 3."
+        maxAttempts := 3
+        if o != nil && o.Attempts != 0 {
+                maxAttempts = o.Attempts
         }
-        return nil
+        for attempt := 0; attempt < maxAttempts; attempt++ {
+                err := loopBody(attempt >= d.txnFakeRetry)
+                if err != errNope {
+                        return err
+                }
+        }
+        return ds.ErrConcurrentTransaction

[iannucci, 2015/09/11 22:31:11]

Can we return a different error when the retries don't execute enough? E.g.
internal error or something?

[Vadim Sh., 2015/09/14 01:07:20]

Real GAE returns ErrConcurrentTransaction:
https://github.com/golang/appengine/blob/master/datastore/transaction.go#L56

In fact, I'll probably remove errNope, since real GAE doesn't bother to
distinguish between real commit error and callback returning
ErrConcurrentTransaction:
https://github.com/golang/appengine/blob/master/internal/transaction.go#L85

 }