Testing different outcomes for transactions.

internal/sqladapter/database.go

@@ -88,6 +88,7 @@ type database struct {
 
 	name   string
 	sess   *sql.DB
+	sessMu sync.Mutex
 
 	cachedStatements  *cache.Cache
 	cachedCollections *cache.Cache
@@ -106,7 +107,9 @@ func (d *database) Session() *sql.DB {
 
 // BindTx binds a *sql.Tx into *database
 func (d *database) BindTx(t *sql.Tx) error {
+	d.sessMu.Lock()
 	d.baseTx = newTx(t)
+	defer d.sessMu.Unlock()
 	return d.Ping()
 }
 
@@ -130,7 +133,9 @@ func (d *database) Name() string {
 
 // BindSession binds a *sql.DB into *database
 func (d *database) BindSession(sess *sql.DB) error {
+	d.sessMu.Lock()
 	d.sess = sess
+	d.sessMu.Unlock()
 
 	if err := d.Ping(); err != nil {
 		return err
@@ -154,11 +159,13 @@ func (d *database) Ping() error {
 // Close terminates the current database session
 func (d *database) Close() error {
 	defer func() {
+		d.sessMu.Lock()
 		d.sess = nil
 		d.baseTx = nil
+		d.sessMu.Unlock()
 	}()
 	if d.sess != nil {
-		if d.Tx() != nil && !d.Tx().Commited() {
+		if d.Tx() != nil && !d.Tx().Committed() {
 			d.Tx().Rollback()
 		}
 		d.cachedStatements.Clear() // Closes prepared statements as well.

internal/sqladapter/testing/adapter.go.tpl

@@ -4,6 +4,7 @@ import (
 	"database/sql"
 	"flag"
 	"log"
+	"fmt"
 	"math/rand"
 	"os"
 	"strconv"
@@ -929,7 +930,7 @@ func TestTransactionsAndRollback(t *testing.T) {
 	assert.NoError(t, err)
 
 	err = tx.Rollback()
-	assert.Error(t, err, "Already commited")
+	assert.Error(t, err, "Already committed")
 
 	// Let's verify we have 3 rows.
 	artist = sess.Collection("artist")
@@ -1083,38 +1084,103 @@ func TestExhaustConnectionPool(t *testing.T) {
 		return
 	}
 
+	var tMu sync.Mutex
+	tFatal := func(err error) {
+		tMu.Lock()
+		defer tMu.Unlock()
+		t.Fatal(err)
+	}
+
+	tLogf := func(format string, args... interface{}) {
+		tMu.Lock()
+		defer tMu.Unlock()
+		t.Logf(format, args...)
+	}
+
 	sess := mustOpen()
 	defer sess.Close()
 
 	var wg sync.WaitGroup
 	for i := 0; i < 300; i++ {
-		wg.Add(1)
-		t.Logf("Tx %d: Pending", i)
+		tLogf("Tx %d: Pending", i)
 
-		go func(t *testing.T, wg *sync.WaitGroup, i int) {
-			var tx db.Tx
+		wg.Add(1)
+		go func(wg *sync.WaitGroup, i int) {
 			defer wg.Done()
 
-			start := time.Now()
-
 			// Requesting a new transaction session.
+			start := time.Now()
 			tx, err := sess.NewTransaction()
 			if err != nil {
-				t.Fatal(err)
+				tFatal(err)
 			}
-
-			t.Logf("Tx %d: OK (waiting time: %v)", i, time.Now().Sub(start))
+			tLogf("Tx %d: OK (time to connect: %v)", i, time.Now().Sub(start))
 
 			// Let's suppose that we do a bunch of complex stuff and that the
 			// transaction lasts 3 seconds.
 			time.Sleep(time.Second * 3)
 
+			switch i%7 {
+			case 0:
+				var account map[string]interface{}
+				if err := tx.Collection("artist").Find().One(&account); err != nil {
+					tFatal(err)
+				}
+				if err := tx.Commit(); err != nil {
+					tFatal(err)
+				}
+				tLogf("Tx %d: Committed", i)
+			case 1:
+				if _, err := tx.DeleteFrom("artist").Exec(); err != nil {
+					tFatal(err)
+				}
+				if err := tx.Rollback(); err != nil {
+					tFatal(err)
+				}
+				tLogf("Tx %d: Rolled back", i)
+			case 2:
 				if err := tx.Close(); err != nil {
-				t.Fatal(err)
+					tFatal(err)
 				}
-
-			t.Logf("Tx %d: Done", i)
-		}(t, &wg, i)
+				tLogf("Tx %d: Closed", i)
+			case 3:
+				var account map[string]interface{}
+				if err := tx.Collection("artist").Find().One(&account); err != nil {
+					tFatal(err)
+				}
+				if err := tx.Commit(); err != nil {
+					tFatal(err)
+				}
+				if err := tx.Close(); err != nil {
+					tFatal(err)
+				}
+				tLogf("Tx %d: Committed and closed", i)
+			case 4:
+				if err := tx.Rollback(); err != nil {
+					tFatal(err)
+				}
+				if err := tx.Close(); err != nil {
+					tFatal(err)
+				}
+				tLogf("Tx %d: Rolled back and closed", i)
+			case 5:
+				if err := tx.Close(); err != nil {
+					tFatal(err)
+				}
+				if err := tx.Commit(); err == nil {
+					tFatal(fmt.Errorf("Error expected"))
+				}
+				tLogf("Tx %d: Closed and committed", i)
+			case 6:
+				if err := tx.Close(); err != nil {
+					tFatal(err)
+				}
+				if err := tx.Rollback(); err == nil {
+					tFatal(fmt.Errorf("Error expected"))
+				}
+				tLogf("Tx %d: Closed and rolled back", i)
+			}
+		}(&wg, i)
 	}
 
 	wg.Wait()

internal/sqladapter/tx.go

@@ -36,7 +36,7 @@ type Tx interface {
 type BaseTx interface {
 	Commit() error
 	Rollback() error
-	Commited() bool
+	Committed() bool
 }
 
 type txWrapper struct {
@@ -61,16 +61,16 @@ func newTxWrapper(db Database) Tx {
 
 type sqlTx struct {
 	*sql.Tx
-	commited atomic.Value
+	committed atomic.Value
 }
 
 func newTx(tx *sql.Tx) BaseTx {
 	return &sqlTx{Tx: tx}
 }
 
-func (t *sqlTx) Commited() bool {
-	commited := t.commited.Load()
-	if commited != nil {
+func (t *sqlTx) Committed() bool {
+	committed := t.committed.Load()
+	if committed != nil {
 		return true
 	}
 	return false
@@ -78,11 +78,21 @@ func (t *sqlTx) Commited() bool {
 
 func (t *sqlTx) Commit() (err error) {
 	if err = t.Tx.Commit(); err == nil {
-		t.commited.Store(struct{}{})
+		t.committed.Store(struct{}{})
 	}
 	return err
 }
 
+func (t *txWrapper) Commit() error {
+	defer t.Database.Close()
+	return t.BaseTx.Commit()
+}
+
+func (t *txWrapper) Rollback() error {
+	defer t.Database.Close()
+	return t.BaseTx.Rollback()
+}
+
 var (
 	_ = BaseTx(&sqlTx{})
 )