memtx: fix a bug in TX that caused deletion of a durty tuple

There was a mess in tuple refernce in TX history.
Now it was remade in the following asumptions:
 * a clean tuple belongs to space, and the space implicitly holds
a reference to the tuple.
 * a dirty tuple belongs to TX manager and a reference is held
in the corresponding story.

Closes #5423

src/box/memtx_space.c

@@ -78,6 +78,7 @@ void
 memtx_space_update_bsize(struct space *space, struct tuple *old_tuple,
 			 struct tuple *new_tuple)
 {
+	assert(space->vtab->destroy == &memtx_space_destroy);
 	struct memtx_space *memtx_space = (struct memtx_space *)space;
 	ssize_t old_bsize = old_tuple ? box_tuple_bsize(old_tuple) : 0;
 	ssize_t new_bsize = new_tuple ? box_tuple_bsize(new_tuple) : 0;

src/box/memtx_tx.c

@@ -418,8 +418,8 @@ memtx_tx_story_gc_step()
 		if (link->newer_story == NULL) {
 			/*
 			 * We are at the top of the chain. That means
-			 * that story->tuple is in index. If the story is
-			 * actually delete the tuple, it must be deleted from
+			 * that story->tuple is in index. If the story
+			 * actually deletes the tuple, it must be deleted from
 			 * index.
 			 */
 			if (story->del_psn > 0 && story->space != NULL) {
@@ -432,6 +432,12 @@ memtx_tx_story_gc_step()
 					panic("failed to rollback change");
 				}
 				assert(story->tuple == unused);
+			} else if (i == 0) {
+				/*
+				 * The tuple is left clean in the space.
+				 * It now belongs to the space and must be referenced.
+				 */
+				tuple_ref(story->tuple);
 			}
 
 			if (link->older.is_story) {
@@ -655,6 +661,14 @@ memtx_tx_history_add_stmt(struct txn_stmt *stmt, struct tuple *old_tuple,
 					  &replaced) != 0)
 				goto fail;
 			memtx_tx_story_link_tuple(add_story, replaced, i);
+			if (i == 0 && replaced != NULL && !replaced->is_dirty) {
+				/*
+				 * The tuple was clean and thus belonged to
+				 * the space. Now tx manager takes ownership
+				 * of it.
+				 */
+				tuple_unref(replaced);
+			}
 			add_story_linked++;
 
 			struct tuple *visible_replaced = NULL;
@@ -729,15 +743,16 @@ memtx_tx_history_add_stmt(struct txn_stmt *stmt, struct tuple *old_tuple,
 		collected_conflicts = collected_conflicts->next;
 	}
 
-	/*
-	 * We now reference both new and old tuple because the stmt holds
-	 * pointers to them.
-	 */
-	if (stmt->new_tuple != NULL)
-		tuple_ref(stmt->new_tuple);
 	*result = old_tuple;
-	if (*result != NULL)
+	if (*result != NULL) {
+		/*
+		 * The result must be a referenced pointer. The caller must
+		 * unreference it by itself.
+		 * Actually now it goes only to stmt->old_tuple, and
+		 * stmt->old_tuple is unreferenced when stmt is destroyed.
+		 */
 		tuple_ref(*result);
+	}
 	return 0;
 
 	fail:
@@ -756,6 +771,10 @@ memtx_tx_history_add_stmt(struct txn_stmt *stmt, struct tuple *old_tuple,
 				unreachable();
 				panic("failed to rollback change");
 			}
+			if (i == 0 && was != NULL && !was->is_dirty) {
+				/* Just rollback previous tuple_unref. */
+				tuple_ref(was);
+			}
 
 			memtx_tx_story_unlink(stmt->add_story, i);
 
@@ -787,6 +806,10 @@ memtx_tx_history_rollback_stmt(struct txn_stmt *stmt)
 		for (uint32_t i = 0; i < story->index_count; i++) {
 			struct memtx_story_link *link = &story->link[i];
 			if (link->newer_story == NULL) {
+				/*
+				 * We are at top of story list and thus
+				 * story->tuple is in index directly.
+				 */
 				struct tuple *unused;
 				struct index *index = stmt->space->index[i];
 				struct tuple *was = memtx_tx_story_older_tuple(link);
@@ -796,6 +819,17 @@ memtx_tx_history_rollback_stmt(struct txn_stmt *stmt)
 					unreachable();
 					panic("failed to rollback change");
 				}
+				if (i == 0 && was != NULL &&
+				    !link->older.is_story) {
+					/*
+					 * That was the last story in history.
+					 * The last tuple now belongs to space
+					 * and the space must hold a reference
+					 * to it.
+					 */
+					tuple_ref(was);
+				}
+
 			} else {
 				struct memtx_story *newer = link->newer_story;
 				assert(newer->link[i].older.is_story);
@@ -813,7 +847,6 @@ memtx_tx_history_rollback_stmt(struct txn_stmt *stmt)
 		}
 		stmt->add_story->add_stmt = NULL;
 		stmt->add_story = NULL;
-		tuple_unref(stmt->new_tuple);
 	}
 
 	if (stmt->del_story != NULL) {
@@ -987,7 +1020,6 @@ memtx_tx_history_commit_stmt(struct txn_stmt *stmt)
 		assert(stmt->del_story->del_stmt == stmt);
 		assert(stmt->next_in_del_list == NULL);
 		res -= stmt->del_story->tuple->bsize;
-		tuple_unref(stmt->del_story->tuple);
 		stmt->del_story->del_stmt = NULL;
 		stmt->del_story = NULL;
 	}

test/box/tx_man.result

@@ -826,6 +826,49 @@ s:select{}
  | - - [1, 3]
  | ...
 
+s:drop()
+ | ---
+ | ...
+s2:drop()
+ | ---
+ | ...
+
+-- https://github.com/tarantool/tarantool/issues/5423
+s = box.schema.space.create('test')
+ | ---
+ | ...
+i1 = s:create_index('pk', {parts={{1, 'uint'}}})
+ | ---
+ | ...
+i2 = s:create_index('sec', {parts={{2, 'uint'}}})
+ | ---
+ | ...
+
+s:replace{1, 0}
+ | ---
+ | - [1, 0]
+ | ...
+s:delete{1}
+ | ---
+ | - [1, 0]
+ | ...
+collectgarbage()
+ | ---
+ | - 0
+ | ...
+s:replace{1, 1}
+ | ---
+ | - [1, 1]
+ | ...
+s:replace{1, 2 }
+ | ---
+ | - [1, 2]
+ | ...
+
+s:drop()
+ | ---
+ | ...
+
 test_run:cmd("switch default")
  | ---
  | - true

test/box/tx_man.test.lua

@@ -236,6 +236,22 @@ s:select{}
 tx3:commit()
 s:select{}
 
+s:drop()
+s2:drop()
+
+-- https://github.com/tarantool/tarantool/issues/5423
+s = box.schema.space.create('test')
+i1 = s:create_index('pk', {parts={{1, 'uint'}}})
+i2 = s:create_index('sec', {parts={{2, 'uint'}}})
+
+s:replace{1, 0}
+s:delete{1}
+collectgarbage()
+s:replace{1, 1}
+s:replace{1, 2 }
+
+s:drop()
+
 test_run:cmd("switch default")
 test_run:cmd("stop server tx_man")
 test_run:cmd("cleanup server tx_man")