From 016e5e0774eb42e73b3b232cf1d875016bdbcbb3 Mon Sep 17 00:00:00 2001
From: Jason Aten <jason.aten@molecula.com>
Date: Mon, 5 Oct 2020 21:08:14 -0500
Subject: [PATCH] no automatic reuse of Qcx

Per slack discussion with Seebs and Nia,
we'll try not automatically resetting the Qcx.

The worry was that our goroutine shutdown
management is so poor that we are asking for
GetTx on a goroutine that still has a Qcx
from a query that was cancelled.

If this is the case, we will now panic instead of
issuing a new Tx. Then we can fix the poor
goroutine management.

 - also require Qcx.Finish or Abort before Reset
---
 field_internal_test.go     |  5 +++++
 txfactory.go               | 29 +++++++++++++++++++++++------
 txfactory_internal_test.go |  2 ++
 3 files changed, 30 insertions(+), 6 deletions(-)

diff --git a/field_internal_test.go b/field_internal_test.go
index deea358e4..d187d5ab9 100644
--- a/field_internal_test.go
+++ b/field_internal_test.go
@@ -609,12 +609,14 @@ func TestBSIGroup_importValue(t *testing.T) {
 			t.Fatalf("test %d, importing values: %s", i, err.Error())
 		}
 		panicOn(qcx.Finish())
+		qcx.Reset()
 		if row, err := f.Range(qcx, f.name, pql.EQ, tt.checkVal); err != nil {
 			t.Fatalf("test %d, getting range: %s", i, err.Error())
 		} else if !reflect.DeepEqual(row.Columns(), tt.expCols) {
 			t.Fatalf("test %d, expected columns: %v, but got: %v", i, tt.expCols, row.Columns())
 		}
 		panicOn(qcx.Finish())
+		qcx.Reset()
 	} // loop
 }
 
@@ -679,6 +681,7 @@ func TestIntField_MinMaxForShard(t *testing.T) {
 				t.Fatalf("test %d, importing values: %s", i, err.Error())
 			}
 			panicOn(qcx.Finish())
+			qcx.Reset()
 
 			shard := uint64(0)
 			tx := f.idx.holder.txf.NewTx(Txo{Write: !writable, Index: f.idx, Field: f.Field, Shard: shard})
@@ -905,6 +908,7 @@ func TestBSIGroup_TxReopenDB(t *testing.T) {
 			t.Fatalf("test %d, importing values: %s", i, err.Error())
 		}
 		panicOn(qcx.Finish())
+		qcx.Reset()
 
 		if row, err := f.Range(qcx, f.name, pql.EQ, tt.checkVal); err != nil {
 			t.Fatalf("test %d, getting range: %s", i, err.Error())
@@ -912,6 +916,7 @@ func TestBSIGroup_TxReopenDB(t *testing.T) {
 			t.Fatalf("test %d, expected columns: %v, but got: %v", i, tt.expCols, row.Columns())
 		}
 		panicOn(qcx.Finish())
+		qcx.Reset()
 	} // loop
 
 	// the test: can we re-open a BSI fragment under Tx store
diff --git a/txfactory.go b/txfactory.go
index e4a868806..4fc489b53 100644
--- a/txfactory.go
+++ b/txfactory.go
@@ -138,6 +138,9 @@ type Qcx struct {
 	// writable tx for all reads and writes on each given
 	// shard
 	write bool
+
+	// don't allow automatic reuse now. Must manually call Reset, or NewQcx().
+	done bool
 }
 
 // Finish commits/rollsback all stored Tx and resets the
@@ -153,7 +156,7 @@ func (q *Qcx) Finish() (err error) {
 		}
 	}
 	err2 := q.Grp.FinishGroup()
-	q.reset()
+	q.done = true
 
 	if err != nil {
 		return err
@@ -171,16 +174,25 @@ func (q *Qcx) Abort() {
 	}
 	q.Grp.AbortGroup()
 
-	q.reset()
+	q.done = true
 }
 
-// reset forgets everything are starts fresh with an empty
+// Reset forgets everything are starts fresh with an empty
 // group, ready for use again as if NewQcx() had been called.
-// q.mu must be held
-func (q *Qcx) reset() {
+func (q *Qcx) Reset() {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	if !q.done {
+		panic("must call Qcx.Abort() or Qcx.Finish() before calling Reset().")
+	}
+	q.unprotected_reset()
+}
+
+func (q *Qcx) unprotected_reset() {
 	q.RequiredForAtomicWriteTx = nil
 	q.RequiredTxo = nil
 	q.Grp = q.Txf.NewTxGroup()
+	q.done = false
 }
 
 // NewQcxWithGroup allocates a freshly allocated and empty Grp.
@@ -236,6 +248,11 @@ func (qcx *Qcx) GetTx(o Txo) (tx Tx, finisher func(perr *error)) {
 	qcx.mu.Lock()
 	defer qcx.mu.Unlock()
 
+	if qcx.done {
+		panic(fmt.Sprintf("cannot call GetTx on a Qcx that is already done. "+
+			"Must call Reset() or txf.NewQcx(); stack='%v'", stack()))
+	}
+
 	// Use direct option if set on QCX.
 	if qcx.Direct {
 		o.Direct = true
@@ -861,7 +878,7 @@ func (f *TxFactory) NewTx(o Txo) (txn Tx) {
 
 	tx, err := dbs.NewTx(o.Write, indexName, o)
 	if err != nil {
-		panic(errors.Wrap(err, "rbfDB.NewRBFTx transaction errored"))
+		panic(errors.Wrap(err, "dbs.NewTx transaction errored"))
 	}
 	return tx
 }
diff --git a/txfactory_internal_test.go b/txfactory_internal_test.go
index 26416f95a..293e6fe21 100644
--- a/txfactory_internal_test.go
+++ b/txfactory_internal_test.go
@@ -74,6 +74,7 @@ func Test_TxFactory_Qcx_query_context(t *testing.T) {
 			finisher(nil) // hit the write tx.Commit path
 			// commit the change, and verify it is still there
 			panicOn(qcx.Finish())
+			qcx.Reset()
 
 			tx, finread := qcx.GetTx(Txo{Write: !writable, Index: idx, Fragment: f, Shard: f.shard})
 			if n := f.mustRow(tx, 120).Count(); n != 2 {
@@ -83,6 +84,7 @@ func Test_TxFactory_Qcx_query_context(t *testing.T) {
 			}
 			finread(nil) // no-op on reads that are in a group, so must qcx.Abort() to stop them.
 			qcx.Abort()
+			qcx.Reset()
 		}
 	}
 	N := 1000