vinyl: fix crash in index drop if there is DML request reading from it

A DML request (insert, replace, update) can yield while reading from
the disk in order to check unique constraints. In the meantime the index
can be dropped. The DML request can't crash in this case thanks to
commit d3e1236 ("vinyl: abort affected transactions when space is
removed from cache"), but the DDL operation can because:
 - It unreferences the index in `alter_space_commit`, which may result
   in dropping the LSM tree with `vy_lsm_delete`.
 - `vy_lsm_delete` may yield in `vy_range_tree_free_cb` while waiting
   for disk readers to complete.
 - Yielding in commit triggers isn't allowed (crashes).

We already fixed a similar issue when `index.get` crashed if raced
with index drop, see commit 75f03a5 ("vinyl: fix crash if space is
dropped while space.get is reading from it"). Let's fix this issue in
the same way - by taking a reference to the LSM tree while checking
unique constraints. To do that it's enough to move `vy_lsm_ref` from
`vinyl_index_get` to `vy_get`.

Also, let's replace `vy_slice_wait_pinned` with an assertion checking
that the slice pin count is 0 in `vy_range_tree_free_cb` because
`vy_lsm_delete` must not yield.

Closes #10094

NO_DOC=bug fix

Author: locker

changelogs/unreleased/gh-10094-vy-index-drop-crash-fix.md

@@ -0,0 +1,4 @@
+## bugfix/vinyl
+
+* Fixed a bug when a DDL operation dropping a unique index could crash
+  if performed concurrently with DML requests (gh-10094).

src/box/vinyl.c

@@ -1408,6 +1408,11 @@ vy_get(struct vy_lsm *lsm, struct vy_tx *tx,
 	 * space.index.get({key}).
 	 */
 	assert(tx == NULL || tx->state == VINYL_TX_READY);
+	/*
+	 * Make sure the LSM tree isn't deleted while we are
+	 * reading from it.
+	 */
+	vy_lsm_ref(lsm);
 
 	int rc;
 	struct vy_entry partial, entry;
@@ -1423,15 +1428,15 @@ vy_get(struct vy_lsm *lsm, struct vy_tx *tx,
 		 * Use point lookup for a full key.
 		 */
 		if (tx != NULL && vy_tx_track_point(tx, lsm, key) != 0)
-			return -1;
+			goto fail;
 		if (vy_point_lookup(lsm, tx, rv, key, &partial) != 0)
-			return -1;
+			goto fail;
 		if (lsm->index_id > 0 && partial.stmt != NULL) {
 			rc = vy_get_by_secondary_tuple(lsm, tx, rv,
 						       partial, &entry);
 			tuple_unref(partial.stmt);
 			if (rc != 0)
-				return -1;
+				goto fail;
 		} else {
 			entry = partial;
 		}
@@ -1457,7 +1462,7 @@ vy_get(struct vy_lsm *lsm, struct vy_tx *tx,
 		vy_read_iterator_cache_add(&itr, entry);
 	vy_read_iterator_close(&itr);
 	if (rc != 0)
-		return -1;
+		goto fail;
 out:
 	*result = entry.stmt;
 
@@ -1472,7 +1477,11 @@ vy_get(struct vy_lsm *lsm, struct vy_tx *tx,
 	}
 	if (*result != NULL)
 		vy_stmt_counter_acct_tuple(&lsm->stat.get, *result);
+	vy_lsm_unref(lsm);
 	return 0;
+fail:
+	vy_lsm_unref(lsm);
+	return -1;
 }
 
 /**
@@ -3862,14 +3871,8 @@ vinyl_index_get(struct index *index, const char *key,
 		tx = &tx_autocommit;
 		vy_tx_create(env->xm, tx);
 	}
-	/*
-	 * Make sure the LSM tree isn't deleted while we are
-	 * reading from it.
-	 */
-	vy_lsm_ref(lsm);
 	int rc = vy_get_by_raw_key(lsm, tx, vy_tx_read_view(tx),
 				   key, part_count, ret);
-	vy_lsm_unref(lsm);
 	if (tx == &tx_autocommit)
 		vy_tx_destroy(tx);
 	if (rc != 0)

src/box/vy_lsm.c

@@ -256,7 +256,7 @@ vy_range_tree_free_cb(vy_range_tree_t *t, struct vy_range *range, void *arg)
 	(void)arg;
 	struct vy_slice *slice;
 	rlist_foreach_entry(slice, &range->slices, in_range)
-		vy_slice_wait_pinned(slice);
+		assert(slice->pin_count == 0);
 	vy_range_delete(range);
 	return NULL;
 }

test/vinyl-luatest/gh_10094_index_drop_test.lua

@@ -0,0 +1,44 @@
+local server = require('luatest.server')
+local t = require('luatest')
+
+local g = t.group()
+
+g.before_all(function(cg)
+    t.tarantool.skip_if_not_debug()
+    cg.server = server:new()
+    cg.server:start()
+    cg.server:exec(function()
+        box.cfg{vinyl_cache = 0}
+    end)
+end)
+
+g.after_all(function(cg)
+    cg.server:drop()
+end)
+
+g.after_each(function(cg)
+    cg.server:exec(function()
+        if box.space.test ~= nil then
+            box.space.test:drop()
+        end
+        box.error.injection.set('ERRINJ_VY_READ_PAGE_DELAY', false)
+    end)
+end)
+
+g.test_index_drop = function(cg)
+    cg.server:exec(function()
+        local fiber = require('fiber')
+        local s = box.schema.create_space('test', {engine = 'vinyl'})
+        s:create_index('pk')
+        s:create_index('sk', {parts = {{2, 'unsigned'}}})
+        s:insert{1, 10}
+        box.snapshot()
+        box.error.injection.set('ERRINJ_VY_READ_PAGE_DELAY', true)
+        fiber.create(function()
+            s:insert{2, 10}
+        end)
+        s.index.sk:drop()
+        box.error.injection.set('ERRINJ_VY_READ_PAGE_DELAY', false)
+        t.assert_equals(s:select({}, {fullscan = true}), {{1, 10}})
+    end)
+end