vinyl: fix !vy_tx_is_in_read_view assertion failure in vy_tx_prepare

Commit 4d52199e ("box: fix transaction "read-view" and "conflicted"
states") updated vy_tx_send_to_read_view so that now it aborts all RW
transactions right away instead of sending them to read view and
aborting them on commit. It also updated vy_tx_begin_statement to fail
if a transaction sent to a read view tries to do DML. With all that,
we assume that there cannot possibly be an RW transaction sent to read
view so we have an assertion checking that in vy_tx_commit.

However, this assertion may fail, because a DML statement may yield
on disk read before it writes anything to the write set. If this is
the first statement in a transaction, the transaction is technically
read-only and we will send it to read-view instead of aborting it.
Once it completes the disk read, it will apply the statement and hence
become read-write, breaking our assumption in vy_tx_commit.

Fix this by aborting RW transactions sent to read-view in vy_tx_set.

Follow-up #7240

NO_DOC=bug fix
NO_CHANGELOG=unreleased

src/box/vy_tx.c

@@ -1149,6 +1149,20 @@ vy_tx_set_entry(struct vy_tx *tx, struct vy_lsm *lsm, struct vy_entry entry)
 int
 vy_tx_set(struct vy_tx *tx, struct vy_lsm *lsm, struct tuple *stmt)
 {
+	if (vy_tx_is_in_read_view(tx)) {
+		/*
+		 * If a conflict occurs while the first DML statement in
+		 * a transaction is waiting for a disk read to check key
+		 * uniqueness, the transaction will not be aborted by
+		 * vy_tx_send_to_read_view, because technically it is still
+		 * read-only. So in addition to vy_tx_begin_statement we
+		 * need to abort transactions sent to read view when we
+		 * add a new statement to the write set.
+		 */
+		assert(vy_tx_is_ro(tx));
+		diag_set(ClientError, ER_TRANSACTION_CONFLICT);
+		return -1;
+	}
 	struct vy_entry entry;
 	vy_stmt_foreach_entry(entry, stmt, lsm->cmp_def) {
 		if (vy_tx_set_entry(tx, lsm, entry) != 0)

test/vinyl-luatest/gh_7240_abort_yielding_dml_test.lua

+local misc = require('test.luatest_helpers.misc')
+local server = require('test.luatest_helpers.server')
+local t = require('luatest')
+
+local g = t.group()
+
+g.before_all(function(cg)
+    cg.server = server:new{alias = 'master'}
+    cg.server:start()
+end)
+
+g.after_all(function(cg)
+    cg.server:drop()
+end)
+
+--
+-- Since gh-7240 was fixed, writing transactions are aborted on conflict
+-- immediately instead of being sent to read view and aborted on commit.
+-- This test checks the following corner case:
+-- 1. The first DML statement in a transaction yields reading disk to check
+--    the uniqueness constraint. The transaction is technically read-only at
+--    this point, because it hasn't added any statements to its write set yet.
+-- 2. Another transaction updates the tuple checked by the first transaction.
+--    Since the first transaction is read-only, it isn't aborted, but sent to
+--    read view.
+-- 3. The first transaction completes the uniqueness check successful and
+--    tries to commit its write set. It should be aborted at this point,
+--    because it's in read view.
+--
+g.test_abort_yielding_dml = function(cg)
+    misc.skip_if_not_debug()
+    cg.server:exec(function()
+        local fiber = require('fiber')
+        local t = require('luatest')
+        local s = box.schema.create_space('s', {engine = 'vinyl'})
+        -- Disable bloom filter to enforce disk reads.
+        s:create_index('pk', {bloom_fpr = 1})
+        -- Make a checkpoint to enable disk reads.
+        s:insert{1}
+        box.snapshot()
+        box.error.injection.set('ERRINJ_VY_READ_PAGE_DELAY', true)
+        local ch = fiber.channel(1)
+        fiber.create(function()
+            local ok, err = pcall(s.insert, s, {2, 20})
+            ch:put(ok or tostring(err))
+        end)
+        -- The insert operation blocks on disk read to check uniqueness.
+        t.assert_is(ch:get(0.1), nil)
+        -- Abort the insert operation blocked on disk read by conflict.
+        s:replace({2, 200})
+        box.error.injection.set('ERRINJ_VY_READ_PAGE_DELAY', false)
+        t.assert_equals(ch:get(0.1), 'Transaction has been aborted by conflict')
+        t.assert_equals(s:get(2), {2, 200})
+        s:drop()
+    end)
+end