limbo: fix commit/rollback failures with triggers

Currently any transaction on synchronous space fails to complete with
the ER_CURSOR_NO_TRANSACTION error, when on_rollback/on_commit triggers
are set. This is caused due to the fact, that some rollback/commit
triggers require in_txn fiber variable to be set but it's not done
when a transaction is completed from the limbo.

Let's assign transaction to the fiber when we complete transaction from
the limbo. Moreover, let's add assertions, which check whether in_txn()
is set, when on_rollback/on_commit triggers are run.

Closes tarantool#8505

NO_DOC=bugfix

changelogs/unreleased/gh-8505-synchro-triggers-fail.md

@@ -0,0 +1,5 @@
+## bugfix/core
+
+* Fixed a bug because of which transactions on synchronous spaces failed with
+  the ER_CURSOR_NO_TRANSACTION error, when on_commit/on_rollback triggers were
+  set (gh-8505).

src/box/txn.c

@@ -764,6 +764,7 @@ txn_complete_fail(struct txn *txn)
 	if (txn->engine != NULL)
 		engine_rollback(txn->engine, txn);
 	if (txn_has_flag(txn, TXN_HAS_TRIGGERS)) {
+		assert(in_txn() != NULL);
 		if (trigger_run(&txn->on_rollback, txn) != 0) {
 			diag_log();
 			panic("transaction rollback trigger failed");
@@ -787,6 +788,7 @@ txn_complete_success(struct txn *txn)
 	if (txn->engine != NULL)
 		engine_commit(txn->engine, txn);
 	if (txn_has_flag(txn, TXN_HAS_TRIGGERS)) {
+		assert(in_txn() != NULL);
 		/*
 		 * Commit triggers must be run in the same order they were added
 		 * so that a trigger sees the changes done by previous triggers

src/box/txn_limbo.c

@@ -72,6 +72,21 @@ txn_limbo_is_ro(struct txn_limbo *limbo)
 	       (limbo->owner_id != instance_id || txn_limbo_is_frozen(limbo));
 }
 
+void
+txn_limbo_complete(struct txn *txn, bool is_success)
+{
+	/*
+	 * Some rollback/commit triggers require the in_txn fiber
+	 * variable to be set.
+	 */
+	fiber_set_txn(fiber(), txn);
+	if (is_success)
+		txn_complete_success(txn);
+	else
+		txn_complete_fail(txn);
+	fiber_set_txn(fiber(), NULL);
+}
+
 struct txn_limbo_entry *
 txn_limbo_last_synchro_entry(struct txn_limbo *limbo)
 {
@@ -287,7 +302,7 @@ txn_limbo_wait_complete(struct txn_limbo *limbo, struct txn_limbo_entry *entry)
 		e->txn->limbo_entry = NULL;
 		txn_limbo_abort(limbo, e);
 		txn_clear_flags(e->txn, TXN_WAIT_SYNC | TXN_WAIT_ACK);
-		txn_complete_fail(e->txn);
+		txn_limbo_complete(e->txn, false);
 		if (e == entry)
 			break;
 		fiber_wakeup(e->txn->fiber);
@@ -461,7 +476,7 @@ txn_limbo_read_confirm(struct txn_limbo *limbo, int64_t lsn)
 		 * after the affected transactions.
 		 */
 		assert(e->txn->signature >= 0);
-		txn_complete_success(e->txn);
+		txn_limbo_complete(e->txn, true);
 	}
 	/*
 	 * Track CONFIRM lsn on replica in order to detect split-brain by
@@ -514,7 +529,7 @@ txn_limbo_read_rollback(struct txn_limbo *limbo, int64_t lsn)
 		assert(e->txn->signature >= 0);
 		e->txn->signature = TXN_SIGNATURE_SYNC_ROLLBACK;
 		e->txn->limbo_entry = NULL;
-		txn_complete_fail(e->txn);
+		txn_limbo_complete(e->txn, false);
 		if (e == last_rollback)
 			break;
 	}

test/box-luatest/gh_8505_synchro_triggers_test.lua

@@ -0,0 +1,66 @@
+local t = require('luatest')
+local replica_set = require('luatest.replica_set')
+local server = require('luatest.server')
+
+local g = t.group('gh-8505-synchro-triggers')
+
+g.before_all(function(g)
+    g.replica_set = replica_set:new({})
+    g.box_cfg = {
+        replication_timeout = 0.01,
+        replication = {
+            server.build_listen_uri('server1', g.replica_set.id),
+            server.build_listen_uri('server2', g.replica_set.id),
+        },
+    }
+
+    g.box_cfg.election_mode = 'voter'
+    g.server2 = g.replica_set:build_and_add_server({
+        alias = 'server2', box_cfg = g.box_cfg
+    })
+
+    g.box_cfg.election_mode = 'candidate'
+    g.server1 = g.replica_set:build_and_add_server({
+        alias = 'server1', box_cfg = g.box_cfg
+    })
+
+    g.replica_set:start()
+    g.server1:wait_for_election_leader()
+    g.server1:exec(function()
+        box.schema.create_space('test', {is_sync = true}):create_index('pk')
+    end)
+    g.server2:wait_for_vclock_of(g.server1)
+end)
+
+g.after_all(function(g)
+    g.replica_set:drop()
+end)
+
+g.test_on_commit_trigger = function(g)
+    g.server1:exec(function()
+        box.begin()
+        box.on_commit(function(iter) iter() end)
+        box.space.test:upsert({1}, {{'=', 1, 1}})
+        box.commit()
+    end)
+end
+
+g.test_on_rollback_trigger = function(g)
+    -- Force ACK gathering to fail and cause rollback
+    local cfg = g.box_cfg
+    cfg.replication_synchro_timeout = 1e-9
+    cfg.election_mode = 'candidate'
+
+    g.server1:update_box_cfg(cfg)
+    g.server1:wait_for_election_leader()
+
+    g.server1:exec(function()
+        box.begin()
+        box.on_rollback(function(iter) iter() end)
+        box.space.test:upsert({1}, {{'=', 1, 1}})
+        local _, err = pcall(box.commit)
+        t.assert_equals(err.code, box.error.SYNC_QUORUM_TIMEOUT)
+    end)
+
+    g.server1:update_box_cfg(g.box_cfg)
+end