memtx: fix lost gap and full scan items

By a mistake in 8a56514 a shortcut was added to procedure
that handles gap write: it was considered that if the writing
transaction is the same as reading - there is no actual conflict
that must be stored further. That was a wrong decision: if such
a transaction yields and another transaction comes and commits
a value with the same key - the first one must go to conflicted
state since it has read no more possible state.

Another similar mistake was made in e6f5090, where writing
after full scan of the same transaction was not tracked as read.
Obviously that was wrong: if some other transaction overwrites
the key and commits - this transaction must go to read view since
it did not see anything by this key which is not so anymore.

Fix it, reverting the first commit and an modifying the second and
add a test.

Closes tarantool#8326

NO_DOC=bugfix

changelogs/unreleased/gh-8326-mvcc-lost-gap-record.md

@@ -0,0 +1,3 @@
+## bugfix/core
+
+* Fixed a bug when MVCC sometimes lost gap record (gh-8326).

src/box/memtx_tx.c

@@ -1774,7 +1774,7 @@ memtx_tx_gap_item_new(struct txn *txn, enum iterator_type type,
  * have read from this gap and thus must be sent to read view or conflicted.
  */
 static void
-memtx_tx_handle_gap_write(struct txn *txn, struct space *space,
+memtx_tx_handle_gap_write(struct space *space,
 			  struct memtx_story *story, struct tuple *tuple,
 			  struct tuple *successor, uint32_t ind)
 {
@@ -1783,9 +1783,8 @@ memtx_tx_handle_gap_write(struct txn *txn, struct space *space,
 	struct rlist *fsc_list = &index->full_scans;
 	uint64_t index_mask = 1ull << (ind & 63);
 	rlist_foreach_entry_safe(fsc_item, fsc_list, in_full_scans, fsc_tmp) {
-		if (fsc_item->txn != txn)
-			memtx_tx_track_read_story(fsc_item->txn, space, story,
-						  index_mask);
+		memtx_tx_track_read_story(fsc_item->txn, space, story,
+					  index_mask);
 	}
 	if (successor != NULL && !tuple_has_flag(successor, TUPLE_IS_DIRTY))
 		return; /* no gap records */
@@ -1825,10 +1824,6 @@ memtx_tx_handle_gap_write(struct txn *txn, struct space *space,
 						item->type == ITER_LT)));
 		bool need_track = need_split ||
 				  (is_full_key && cmp == 0 && is_e);
-		/* There's no need to track read of own change. */
-		if (story->add_stmt != NULL &&
-		    story->add_stmt->txn == item->txn)
-			need_track = false;
 		if (need_track)
 			memtx_tx_track_read_story(item->txn, space,
 						  story, index_mask);
@@ -1917,8 +1912,7 @@ memtx_tx_history_add_insert_stmt(struct txn_stmt *stmt,
 		memtx_tx_story_link_top(add_story, replaced_story, 0, true);
 	} else {
 		memtx_tx_story_link_top(add_story, NULL, 0, true);
-		memtx_tx_handle_gap_write(stmt->txn, space,
-					  add_story, new_tuple,
+		memtx_tx_handle_gap_write(space, add_story, new_tuple,
 					  direct_successor[0], 0);
 	}
 
@@ -1929,8 +1923,7 @@ memtx_tx_history_add_insert_stmt(struct txn_stmt *stmt,
 
 	for (uint32_t i = 1; i < space->index_count; i++) {
 		if (directly_replaced[i] == NULL) {
-			memtx_tx_handle_gap_write(stmt->txn, space,
-						  add_story, new_tuple,
+			memtx_tx_handle_gap_write(space, add_story, new_tuple,
 						  direct_successor[i], i);
 			continue;
 		}

test/box-luatest/gh_8326_mvcc_lost_gap_record_test.lua

@@ -0,0 +1,72 @@
+local server = require('luatest.server')
+local t = require('luatest')
+
+local g = t.group()
+
+g.before_all(function()
+    g.server = server:new{
+        alias   = 'default',
+        box_cfg = {memtx_use_mvcc_engine = true}
+    }
+    g.server:start()
+end)
+
+g.after_all(function()
+    g.server:drop()
+end)
+
+g.before_each(function()
+    g.server:exec(function()
+        local s = box.schema.space.create('s')
+        s:create_index('pk', {parts = {{1, 'uint'}, {2, 'uint'}}})
+        s:create_index('sk', {type = 'hash', parts = {{2, 'uint'}}})
+    end)
+end)
+
+g.after_each(function()
+    g.server:exec(function()
+        box.space.s:drop()
+    end)
+end)
+
+g.test_lost_gap_record = function()
+    g.server:exec(function()
+        local txn_proxy = require("test.box.lua.txn_proxy")
+
+        local tx1 = txn_proxy.new()
+        local tx2 = txn_proxy.new()
+        tx1:begin()
+        tx2:begin()
+
+        tx1('box.space.s:select{1}') -- select by partial key {1}, empty result
+        tx1('box.space.s:replace{1, 1, 1}') -- write right to selected
+
+        tx2('box.space.s:replace{1, 1, 2}') -- overwrite by the second TX
+        tx2:commit() -- ok, now tx1 must become conflicted because of select{1}
+
+        t.assert_equals(tx1:commit(),
+                        {{error = "Transaction has been aborted by conflict"}})
+        t.assert_equals(box.space.s:select{1}, {{1, 1, 2}})
+    end)
+end
+
+g.test_lost_full_scan_record = function()
+    g.server:exec(function()
+        local txn_proxy = require("test.box.lua.txn_proxy")
+
+        local tx1 = txn_proxy.new()
+        local tx2 = txn_proxy.new()
+        tx1:begin()
+        tx2:begin()
+
+        tx1('box.space.s.index.sk:select{}') -- secondary fullscan, empty result
+        tx1('box.space.s:replace{1, 1, 1}') -- write to selected
+
+        tx2('box.space.s:replace{1, 1, 2}') -- overwrite by the second TX
+        tx2:commit() -- ok, now tx1 must become conflicted because of select{1}
+
+        t.assert_equals(tx1:commit(),
+                        {{error = "Transaction has been aborted by conflict"}})
+        t.assert_equals(box.space.s:select{1}, {{1, 1, 2}})
+    end)
+end