Revert back to using WriteTxn as argument to Changes()

Using a ReadTxn and then optionally doing a WriteTxn is Changes()
was, in hindsight, very racy. The iterator must be seeded in the same
WriteTxn in which the delete tracker is registered as otherwise we may
end up seeing inserted objects that were deleted and missing the delete
changes.

Signed-off-by: Jussi Maki <[email protected]>

Author: joamaki

benchmarks_test.go

@@ -295,11 +295,13 @@ func BenchmarkDB_Changes(b *testing.B) {
 	b.ResetTimer()
 	for n := 0; n < b.N; n++ {
 		// Create the change iterator.
-		iter, err := table.Changes(db.ReadTxn())
+		txn := db.WriteTxn(table)
+		iter, err := table.Changes(txn)
+		txn.Commit()
 		require.NoError(b, err)
 
 		// Create objects
-		txn := db.WriteTxn(table)
+		txn = db.WriteTxn(table)
 		for i := 0; i < numObjectsToInsert; i++ {
 			_, _, err := table.Insert(txn, testObject{ID: uint64(i), Tags: nil})
 			if err != nil {

db_test.go

@@ -277,11 +277,12 @@ func TestDB_EventIterator(t *testing.T) {
 	assert.EqualValues(t, 0, expvarInt(metrics.GraveyardObjectCountVar.Get("test")), "GraveyardObjectCount")
 
 	// Create two change iterators
-	txn := db.ReadTxn()
-	iter, err := table.Changes(txn)
+	wtxn := db.WriteTxn(table)
+	iter, err := table.Changes(wtxn)
 	require.NoError(t, err, "failed to create ChangeIterator")
-	iter2, err := table.Changes(txn)
+	iter2, err := table.Changes(wtxn)
 	require.NoError(t, err, "failed to create ChangeIterator")
+	wtxn.Commit()
 
 	assert.EqualValues(t, 2, expvarInt(metrics.DeleteTrackerCountVar.Get("test")), "DeleteTrackerCount")
 
@@ -311,7 +312,7 @@ func TestDB_EventIterator(t *testing.T) {
 	}
 
 	// 1 object should exist.
-	txn = db.ReadTxn()
+	txn := db.ReadTxn()
 	iterAll, _ := table.All(txn)
 	objs := Collect(iterAll)
 	require.Len(t, objs, 1)

derive.go

@@ -70,7 +70,9 @@ type derive[In, Out any] struct {
 
 func (d derive[In, Out]) loop(ctx context.Context, health cell.Health) error {
 	out := d.OutTable
-	iter, err := d.InTable.Changes(d.DB.ReadTxn())
+	txn := d.DB.WriteTxn(d.InTable)
+	iter, err := d.InTable.Changes(txn)
+	txn.Commit()
 	if err != nil {
 		return err
 	}

fuzz_test.go

@@ -371,8 +371,9 @@ func randomAction() action {
 
 func trackerWorker(i int, stop <-chan struct{}) {
 	log := newDebugLogger(900 + i)
-	txn := fuzzDB.ReadTxn()
-	iter, err := tableFuzz1.Changes(txn)
+	wtxn := fuzzDB.WriteTxn(tableFuzz1)
+	iter, err := tableFuzz1.Changes(wtxn)
+	wtxn.Commit()
 	if err != nil {
 		panic(err)
 	}
@@ -381,6 +382,7 @@ func trackerWorker(i int, stop <-chan struct{}) {
 	// Keep track of what state the changes lead us to in order to validate it.
 	state := map[uint64]*statedb.Change[fuzzObj]{}
 
+	var txn statedb.ReadTxn
 	var prevRev statedb.Revision
 	for {
 		for change, rev, ok := iter.Next(); ok; change, rev, ok = iter.Next() {
@@ -406,31 +408,34 @@ func trackerWorker(i int, stop <-chan struct{}) {
 				state[change.Object.id] = &change
 			}
 		}
-		// Validate that the observed changes match with the database state at this
-		// snapshot.
-		state2 := maps.Clone(state)
-		iterAll, _ := tableFuzz1.All(txn)
-		for obj, rev, ok := iterAll.Next(); ok; obj, rev, ok = iterAll.Next() {
-			change, found := state[obj.id]
-			if !found {
-				panic(fmt.Sprintf("trackerWorker: object %d not found from state", obj.id))
-			}
 
-			if change.Revision != rev {
-				panic(fmt.Sprintf("trackerWorker: last observed revision %d does not match real revision %d", change.Revision, rev))
-			}
+		if txn != nil {
+			// Validate that the observed changes match with the database state at this
+			// snapshot.
+			state2 := maps.Clone(state)
+			iterAll, _ := tableFuzz1.All(txn)
+			for obj, rev, ok := iterAll.Next(); ok; obj, rev, ok = iterAll.Next() {
+				change, found := state[obj.id]
+				if !found {
+					panic(fmt.Sprintf("trackerWorker: object %d not found from state", obj.id))
+				}
+
+				if change.Revision != rev {
+					panic(fmt.Sprintf("trackerWorker: last observed revision %d does not match real revision %d", change.Revision, rev))
+				}
 
-			if change.Object.value != obj.value {
-				panic(fmt.Sprintf("trackerWorker: observed value %d does not match real value %d", change.Object.value, obj.value))
+				if change.Object.value != obj.value {
+					panic(fmt.Sprintf("trackerWorker: observed value %d does not match real value %d", change.Object.value, obj.value))
+				}
+				delete(state2, obj.id)
 			}
-			delete(state2, obj.id)
-		}
 
-		if len(state) != tableFuzz1.NumObjects(txn) {
-			for id := range state2 {
-				fmt.Printf("%d should not exist\n", id)
+			if len(state) != tableFuzz1.NumObjects(txn) {
+				for id := range state2 {
+					fmt.Printf("%d should not exist\n", id)
+				}
+				panic(fmt.Sprintf("trackerWorker: state size mismatch: %d vs %d", len(state), tableFuzz1.NumObjects(txn)))
 			}
-			panic(fmt.Sprintf("trackerWorker: state size mismatch: %d vs %d", len(state), tableFuzz1.NumObjects(txn)))
 		}
 
 		txn = fuzzDB.ReadTxn()
@@ -524,7 +529,7 @@ func TestDB_Fuzz(t *testing.T) {
 		}()
 		// Delay a bit to start the trackers at different points in time
 		// so they will observe a different starting state.
-		time.Sleep(100 * time.Millisecond)
+		time.Sleep(500 * time.Millisecond)
 	}
 
 	// Wait until the mutation workers stop and then stop

observable.go

@@ -25,7 +25,9 @@ type observable[Obj any] struct {
 
 func (to *observable[Obj]) Observe(ctx context.Context, next func(Change[Obj]), complete func(error)) {
 	go func() {
-		iter, err := to.table.Changes(to.db.ReadTxn())
+		txn := to.db.WriteTxn(to.table)
+		iter, err := to.table.Changes(txn)
+		txn.Commit()
 		if err != nil {
 			complete(err)
 			return

table.go

@@ -318,30 +318,13 @@ func (t *genTable[Obj]) DeleteAll(txn WriteTxn) error {
 	return nil
 }
 
-func (t *genTable[Obj]) Changes(txn ReadTxn) (ChangeIterator[Obj], error) {
+func (t *genTable[Obj]) Changes(txn WriteTxn) (ChangeIterator[Obj], error) {
 	iter := &changeIterator[Obj]{
 		revision: 0,
 		table:    t,
 	}
 
-	// Check that 'txn' is not a 'WriteTxn' against this table.
-	for _, e := range txn.getTxn().modifiedTables {
-		if e != nil && e.meta.Name() == t.table {
-			// 'txn' is a WriteTxn against the same table. Refuse this as the snapshot
-			// to seed the initial iterator from.
-			//
-			// The user might use the same transaction to insert or delete objects
-			// and it might be confusing as these changes may or may not be observed
-			// depending on whether Changes() was called before Insert() or Delete(),
-			// and since this is an unexpected use-case it's better to just reject this.
-			return nil, fmt.Errorf("Changes() cannot be called with a write transaction against table %q", t.table)
-		}
-	}
-
-	// Create a WriteTxn to add the delete tracker.
-	itxn := txn.getTxn().db.WriteTxn(t).getTxn()
-	defer itxn.Commit()
-
+	itxn := txn.getTxn()
 	name := fmt.Sprintf("iterator-%p", iter)
 	iter.dt = &deleteTracker[Obj]{
 		db:          itxn.db,

txn.go

@@ -242,6 +242,10 @@ func (txn *txn) addDeleteTracker(meta TableMeta, trackerName string, dt anyDelet
 		return ErrTransactionClosed
 	}
 	table := txn.modifiedTables[meta.tablePos()]
+	if table == nil {
+		return tableError(meta.Name(), ErrTableNotLockedForWriting)
+	}
+
 	_, _, table.deleteTrackers = table.deleteTrackers.Insert([]byte(trackerName), dt)
 	txn.db.metrics.DeleteTrackerCount(meta.Name(), table.deleteTrackers.Len())
 

types.go

@@ -71,7 +71,7 @@ type Table[Obj any] interface {
 	//
 	// If an object is created and deleted before the observer has iterated
 	// over the creation then only the deletion is seen.
-	Changes(txn ReadTxn) (ChangeIterator[Obj], error)
+	Changes(WriteTxn) (ChangeIterator[Obj], error)
 }
 
 // Change is either an update or a delete of an object. Used by Changes() and