[PATCH] txn: fix rollback in sub statement

[Vladimir Davydov]

There are two issues in the rollback code:

 - txn_rollback_stmt() rollbacks the current autocommit transaction even
   if it is called from a sub-statement. As a result, if a sub-statement
   (i.e. a statement called from a before_replace or on_replace trigger)
   fails (e.g. due to a conflict), it will trash the current transaction
   leading to a bad memory access upon returning from the trigger.

 - txn_begin_stmt() calls txn_rollback_stmt() on failure even if it did
   not instantiate the statement. So if it is called from a trigger and
   fails (e.g. due to nesting limit), it may trash the parent statement,
   again leading to a crash.

Fix them both and add some tests.

Closes #3127
---
Branch: gh-3127-txn-fix-rollback-in-sub-stmt

 src/box/txn.c                    | 22 ++++++++--------
 test/box/before_replace.result   | 57 +++++++++++++++++++++++++++++++++++++++-
 test/box/before_replace.test.lua | 21 ++++++++++++++-
 3 files changed, 87 insertions(+), 13 deletions(-)

diff --git a/src/box/txn.c b/src/box/txn.c
index 1af8243e..e25c0e0e 100644
--- a/src/box/txn.c
+++ b/src/box/txn.c
@@ -166,10 +166,10 @@ txn_begin_stmt(struct space *space)
 	if (txn == NULL) {
 		txn = txn_begin(true);
 		if (txn == NULL)
-			goto fail;
+			return NULL;
 	} else if (txn->in_sub_stmt > TXN_SUB_STMT_MAX) {
 		diag_set(ClientError, ER_SUB_STMT_MAX);
-		goto fail;
+		return NULL;
 	}
 
 	if (trigger_run(&space->on_stmt_begin, txn) != 0)
@@ -184,12 +184,14 @@ txn_begin_stmt(struct space *space)
 		goto fail;
 	stmt->space = space;
 
-	if (engine_begin_statement(engine, txn) != 0)
-		goto fail;
-
+	if (engine_begin_statement(engine, txn) != 0) {
+		txn_rollback_stmt();
+		return NULL;
+	}
 	return txn;
 fail:
-	txn_rollback_stmt();
+	if (txn->is_autocommit && txn->in_sub_stmt == 0)
+		txn_rollback();
 	return NULL;
 }
 
@@ -325,13 +327,11 @@ void
 txn_rollback_stmt()
 {
 	struct txn *txn = in_txn();
-	if (txn == NULL)
-		return;
-	if (txn->is_autocommit)
-		return txn_rollback();
-	if (txn->in_sub_stmt == 0)
+	if (txn == NULL || txn->in_sub_stmt == 0)
 		return;
 	txn->in_sub_stmt--;
+	if (txn->is_autocommit && txn->in_sub_stmt == 0)
+		return txn_rollback();
 	txn_rollback_to_svp(txn, txn->sub_stmt_begin[txn->in_sub_stmt]);
 }
 
diff --git a/test/box/before_replace.result b/test/box/before_replace.result
index 286b0c59..10f4699c 100644
--- a/test/box/before_replace.result
+++ b/test/box/before_replace.result
@@ -405,21 +405,31 @@ s2 = box.schema.space.create('test2')
 _ = s2:create_index('pk')
 ---
 ...
-cb = function(old, new) s2:auto_increment{old, new} end
+cb = function(old, new) s2:insert{i, old, new} end
 ---
 ...
 s:before_replace(cb) == cb
 ---
 - true
 ...
+i = 1
+---
+...
 s:insert{1, 1}
 ---
 - [1, 1]
 ...
+i = 2
+---
+...
 s:replace{1, 2}
 ---
 - [1, 2]
 ...
+s:replace{1, 3} -- error: conflict in s2
+---
+- error: Duplicate key exists in unique index 'pk' in space 'test2'
+...
 s:select()
 ---
 - - [1, 2]
@@ -568,6 +578,51 @@ s:delete(1)
 ---
 - [1, 2, 3]
 ...
+-- Nesting limit: space.before_replace
+cb = function(old, new) s:insert{1, 1} end
+---
+...
+s:before_replace(cb) == cb
+---
+- true
+...
+s:insert{1, 1} -- error
+---
+- error: 'Can not execute a nested statement: nesting limit reached'
+...
+s:select()
+---
+- []
+...
+s:before_replace(nil, cb)
+---
+...
+-- Nesting limit: space.before_replace + space.on_replace
+cb = function(old, new) s:delete(1) end
+---
+...
+s:before_replace(cb) == cb
+---
+- true
+...
+s:on_replace(cb) == cb
+---
+- true
+...
+s:insert{1, 1} -- error
+---
+- error: 'Can not execute a nested statement: nesting limit reached'
+...
+s:select()
+---
+- []
+...
+s:before_replace(nil, cb)
+---
+...
+s:on_replace(nil, cb)
+---
+...
 -- Make sure the server can recover from xlogs after
 -- using space:before_replace.
 test_run:cmd('restart server default')
diff --git a/test/box/before_replace.test.lua b/test/box/before_replace.test.lua
index 7514baaf..01a386d9 100644
--- a/test/box/before_replace.test.lua
+++ b/test/box/before_replace.test.lua
@@ -132,10 +132,13 @@ s:delete(1)
 -- Issue DML from trigger.
 s2 = box.schema.space.create('test2')
 _ = s2:create_index('pk')
-cb = function(old, new) s2:auto_increment{old, new} end
+cb = function(old, new) s2:insert{i, old, new} end
 s:before_replace(cb) == cb
+i = 1
 s:insert{1, 1}
+i = 2
 s:replace{1, 2}
+s:replace{1, 3} -- error: conflict in s2
 s:select()
 s2:select()
 
@@ -184,6 +187,22 @@ s:before_replace(nil, ret_update)
 s:on_replace(nil, save)
 s:delete(1)
 
+-- Nesting limit: space.before_replace
+cb = function(old, new) s:insert{1, 1} end
+s:before_replace(cb) == cb
+s:insert{1, 1} -- error
+s:select()
+s:before_replace(nil, cb)
+
+-- Nesting limit: space.before_replace + space.on_replace
+cb = function(old, new) s:delete(1) end
+s:before_replace(cb) == cb
+s:on_replace(cb) == cb
+s:insert{1, 1} -- error
+s:select()
+s:before_replace(nil, cb)
+s:on_replace(nil, cb)
+
 -- Make sure the server can recover from xlogs after
 -- using space:before_replace.
 test_run:cmd('restart server default')
-- 
2.11.0