From 439b56819ab3071c0725031d1be652777945329b Mon Sep 17 00:00:00 2001
From: Roy Crihfield <roy@manteia.ltd>
Date: Thu, 8 Jul 2021 15:08:30 +0800
Subject: [PATCH] fix: Fix conflict detection for managed DB (#1716)

I propose this simple fix for detecting conflicts in managed mode. Addresses https://discuss.dgraph.io/t/fatal-error-when-writing-conflicting-keys-in-managed-mode/14784.

When a write conflict exists for a managed DB, an internal assert can fail.
This occurs because a detected conflict is indicated with commitTs of 0, but handling the error is skipped for managed DB instances.

Rather than conflate conflict detection with a timestamp of 0, it can be indicated with another return value from hasConflict.

(cherry picked from commit 3279e188a512751c2658221e64095312203869cd)
---
 txn.go      | 12 +++++-------
 txn_test.go |  9 +++++++++
 2 files changed, 14 insertions(+), 7 deletions(-)

diff --git a/txn.go b/txn.go
index 519906a3c..7ab42e60d 100644
--- a/txn.go
+++ b/txn.go
@@ -160,12 +160,12 @@ func (o *oracle) hasConflict(txn *Txn) bool {
 	return false
 }
 
-func (o *oracle) newCommitTs(txn *Txn) uint64 {
+func (o *oracle) newCommitTs(txn *Txn) (uint64, bool) {
 	o.Lock()
 	defer o.Unlock()
 
 	if o.hasConflict(txn) {
-		return 0
+		return 0, true
 	}
 
 	var ts uint64
@@ -194,7 +194,7 @@ func (o *oracle) newCommitTs(txn *Txn) uint64 {
 		})
 	}
 
-	return ts
+	return ts, false
 }
 
 func (o *oracle) doneRead(txn *Txn) {
@@ -537,10 +537,8 @@ func (txn *Txn) commitAndSend() (func() error, error) {
 	orc.writeChLock.Lock()
 	defer orc.writeChLock.Unlock()
 
-	commitTs := orc.newCommitTs(txn)
-	// The commitTs can be zero if the transaction is running in managed mode.
-	// Individual entries might have their own timestamps.
-	if commitTs == 0 && !txn.db.opt.managedTxns {
+	commitTs, conflict := orc.newCommitTs(txn)
+	if conflict {
 		return nil, ErrConflict
 	}
 
diff --git a/txn_test.go b/txn_test.go
index 021f74b26..9823a9cff 100644
--- a/txn_test.go
+++ b/txn_test.go
@@ -817,6 +817,15 @@ func TestManagedDB(t *testing.T) {
 			}
 		}
 		txn.Discard()
+
+		// Write data to same key, causing a conflict
+		txn = db.NewTransactionAt(10, true)
+		txnb := db.NewTransactionAt(10, true)
+		txnb.Get(key(0))
+		require.NoError(t, txn.SetEntry(NewEntry(key(0), val(0))))
+		require.NoError(t, txnb.SetEntry(NewEntry(key(0), val(1))))
+		require.NoError(t, txn.CommitAt(11, nil))
+		require.Equal(t, ErrConflict, txnb.CommitAt(11, nil))
 	}
 	t.Run("disk mode", func(t *testing.T) {
 		db, err := Open(opt)