vinyl: fix use-after-free in vy_read_iterator_next

A read source iterator stores statements in a vy_history object using
vy_history_append_stmt(). If a statement can be referenced, it's
reference counter is incremented. If it can't, i.e. it belongs to a
memory source, it's stored in a vy_history object without referencing.

This works fine because memory sources are append-only. A problem arises
only when we get to scanning disk sources. Since we yield while reading
disk, a dump task may complete concurrently dropping the memory sources
and possibly invalidating statements stored in the iterator history.
Although we drop the history accumulated so far and restart the
iteration from scratch in this case, there's still an issue that can
result in a use-after-free bug in vy_read_iterator_next().

The problem is that we access the current candidate for the next
statement while evaluating a disk source after a disk read. If 'next'
refers to a referenced statement, it's fine, but if it refers to a
statement from a memory source, it may cause use-after-free because
the memory source may be dropped during a disk read.

To fix this issue, let's make vy_history_append_stmt() copy statements
coming from memory sources. This should be fine performance-wise because
we copied memory statements eventually in vy_history_apply() anyway,
before returning them to the user.

Note that we also have to update vy_read_iterator_restore_mem() because
it implicitly relied on the fact that 'next' coming from a memory source
can't be freed by vy_mem_iterator_restore(), which cleans up the memory
source history. Now, it isn't true anymore so we have to temporarily
take a reference to 'next' explicitly.

Closes tarantool#8852

NO_DOC=bug fix
NO_TEST=tested by ASAN

changelogs/unreleased/gh-8852-vy-read-iterator-use-after-free-fix.md

@@ -0,0 +1,5 @@
+## bugfix/vinyl
+
+* Fixed a heap-use-after-free bug in the Vinyl read iterator caused by a race
+  between a disk read and a memory dump task. The bug could lead to a crash or
+  an invalid query result (gh-8852).

src/box/vy_history.c

@@ -52,9 +52,15 @@ vy_history_append_stmt(struct vy_history *history, struct vy_entry entry)
 			 "struct vy_history_node");
 		return -1;
 	}
-	node->is_refable = vy_stmt_is_refable(entry.stmt);
-	if (node->is_refable)
+	if (vy_stmt_is_refable(entry.stmt)) {
 		tuple_ref(entry.stmt);
+	} else {
+		entry.stmt = vy_stmt_dup(entry.stmt);
+		if (entry.stmt == NULL) {
+			mempool_free(history->pool, node);
+			return -1;
+		}
+	}
 	node->entry = entry;
 	rlist_add_tail_entry(&history->stmts, node, link);
 	return 0;
@@ -65,8 +71,7 @@ vy_history_cleanup(struct vy_history *history)
 {
 	struct vy_history_node *node, *tmp;
 	rlist_foreach_entry_safe(node, &history->stmts, link, tmp) {
-		if (node->is_refable)
-			tuple_unref(node->entry.stmt);
+		tuple_unref(node->entry.stmt);
 		mempool_free(history->pool, node);
 	}
 	rlist_create(&history->stmts);
@@ -91,11 +96,6 @@ vy_history_apply(struct vy_history *history, struct key_def *cmp_def,
 			 * Ignore terminal delete unless the caller
 			 * explicitly asked to keep it.
 			 */
-		} else if (!node->is_refable) {
-			curr.hint = node->entry.hint;
-			curr.stmt = vy_stmt_dup(node->entry.stmt);
-			if (curr.stmt == NULL)
-				return -1;
 		} else {
 			curr = node->entry;
 			tuple_ref(curr.stmt);

src/box/vy_history.h

@@ -59,22 +59,8 @@ struct vy_history {
 struct vy_history_node {
 	/** Link in a history list. */
 	struct rlist link;
-	/** History statement. Referenced if @is_refable is set. */
+	/** History statement. Always referenced. */
 	struct vy_entry entry;
-	/**
-	 * Set if the statement stored in this node is refable,
-	 * i.e. has a reference counter that can be incremented
-	 * to pin the statement in memory. Refable statements are
-	 * referenced by the history. It is a responsibility of
-	 * the user of the history to track lifetime of unrefable
-	 * statements.
-	 *
-	 * Note, we need to store this flag here, because by the
-	 * time we clean up a history list, unrefable statements
-	 * stored in it might have been deleted, thus making
-	 * vy_stmt_is_refable() unusable.
-	 */
-	bool is_refable;
 };
 
 /**

src/box/vy_read_iterator.c

@@ -418,13 +418,24 @@ vy_read_iterator_restore_mem(struct vy_read_iterator *itr,
 	int cmp;
 	struct vy_read_src *src = &itr->src[itr->mem_src];
 
+	/*
+	 * 'next' may refer to a statement in the memory source history,
+	 * which may be cleaned up by vy_mem_iterator_restore(), so we need
+	 * to take a reference to it.
+	 */
+	struct tuple *next_stmt_ref = next->stmt;
+	if (next_stmt_ref != NULL)
+		tuple_ref(next_stmt_ref);
+
 	rc = vy_mem_iterator_restore(&src->mem_iterator,
 				     itr->last, &src->history);
 	if (rc < 0)
-		return -1; /* memory allocation error */
+		goto out; /* memory allocation error */
 	if (rc == 0)
-		return 0; /* nothing changed */
+		goto out; /* nothing changed */
 
+	/* The memory source was updated. Reevaluate it for 'next'. */
+	rc = 0;
 	struct vy_entry entry = vy_history_last_stmt(&src->history);
 	cmp = vy_read_iterator_cmp_stmt(itr, entry, *next);
 	if (cmp > 0) {
@@ -439,14 +450,15 @@ vy_read_iterator_restore_mem(struct vy_read_iterator *itr,
 		 */
 		if (src->front_id == itr->front_id)
 			vy_read_iterator_reevaluate_srcs(itr, next);
-		return 0;
+		goto out;
 	}
+	/* The new statement is a better candidate for 'next'. */
+	*next = entry;
 	if (cmp < 0) {
 		/*
 		 * The new statement precedes the current
 		 * candidate for the next key.
 		 */
-		*next = entry;
 		itr->front_id++;
 	} else {
 		/*
@@ -459,7 +471,10 @@ vy_read_iterator_restore_mem(struct vy_read_iterator *itr,
 			vy_history_cleanup(&cache_src->history);
 	}
 	src->front_id = itr->front_id;
-	return 0;
+out:
+	if (next_stmt_ref != NULL)
+		tuple_unref(next_stmt_ref);
+	return rc;
 }
 
 static void