vinyl: fix cache invalidation on rollback of DELETE statement

Once a statement is prepared to be committed to WAL, it becomes visible
(in the 'read-committed' isolation level) so it can be added to the
tuple cache. That's why if the statement is rolled back due to a WAL
error, we have to invalidate the cache. The problem is that the function
invalidating the cache (`vy_cache_on_write`) ignores the statement if
it's a DELETE judging that "there was nothing and there is nothing now".
This is apparently wrong for rollback. Fix it.

Closes #10879

NO_DOC=bug fix

(cherry picked from commit d64e29da2c323a4b4fcc7cf9fddb0300d5dd081f)

changelogs/unreleased/gh-10879-fix-vinyl-cache-invalidation-on-rollback.md

+## bugfix/vinyl
+
+* Fixed a bug when the tuple cache was not properly invalidated in case
+  a WAL write error occurred while committing a `space.delete()` operation.
+  The bug could lead to a crash or an invalid read query result (gh-10879).

src/box/vy_cache.c

@@ -434,9 +434,12 @@ vy_cache_get(struct vy_cache *cache, struct vy_entry key)
 	return (*node)->entry;
 }
 
-void
-vy_cache_on_write(struct vy_cache *cache, struct vy_entry entry,
-		  struct vy_entry *deleted)
+/**
+ * Invalidate the cache when a statement is committed or rolled back.
+ */
+static void
+vy_cache_invalidate(struct vy_cache *cache, struct vy_entry entry,
+		    bool rollback, struct vy_entry *deleted)
 {
 	vy_cache_gc(cache->env);
 	bool exact = false;
@@ -457,7 +460,7 @@ vy_cache_on_write(struct vy_cache *cache, struct vy_entry entry,
 	 *   ('exact' == false, 'node' == NULL)
 	 */
 
-	if (vy_stmt_type(entry.stmt) == IPROTO_DELETE && !exact) {
+	if (!rollback && vy_stmt_type(entry.stmt) == IPROTO_DELETE && !exact) {
 		/* there was nothing and there is nothing now */
 		return;
 	}
@@ -511,6 +514,19 @@ vy_cache_on_write(struct vy_cache *cache, struct vy_entry entry,
 	}
 }
 
+void
+vy_cache_on_write(struct vy_cache *cache, struct vy_entry entry,
+		  struct vy_entry *deleted)
+{
+	vy_cache_invalidate(cache, entry, /*rollback=*/false, deleted);
+}
+
+void
+vy_cache_on_rollback(struct vy_cache *cache, struct vy_entry entry)
+{
+	vy_cache_invalidate(cache, entry, /*rollback=*/true, NULL);
+}
+
 /**
  * Get a stmt by current position
  */

src/box/vy_cache.h

@@ -233,6 +233,14 @@ void
 vy_cache_on_write(struct vy_cache *cache, struct vy_entry entry,
 		  struct vy_entry *deleted);
 
+/**
+ * Invalidate cache on statement rollback.
+ * @param cache - pointer to tuple cache.
+ * @param entry - rolled back statement.
+ */
+void
+vy_cache_on_rollback(struct vy_cache *cache, struct vy_entry entry);
+
 
 /**
  * Cache iterator

src/box/vy_lsm.c

@@ -1031,9 +1031,7 @@ vy_lsm_rollback_stmt(struct vy_lsm *lsm, struct vy_mem *mem,
 		     struct vy_entry entry)
 {
 	vy_mem_rollback_stmt(mem, entry, &lsm->stat.memory.count);
-
-	/* Invalidate cache element. */
-	vy_cache_on_write(&lsm->cache, entry, NULL);
+	vy_cache_on_rollback(&lsm->cache, entry);
 }
 
 int

test/vinyl-luatest/gh_10879_invalidate_cache_on_rollback_test.lua

+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({
+        box_cfg = {
+            vinyl_cache = 1024 * 1024,
+        },
+    })
+    cg.server:start()
+end)
+
+g.after_all(function(cg)
+    cg.server:drop()
+end)
+
+g.after_each(function(cg)
+    cg.server:exec(function()
+        box.error.injection.set('ERRINJ_WAL_WRITE', false)
+        box.error.injection.set('ERRINJ_WAL_DELAY', false)
+        if box.space.test ~= nil then
+            box.space.test:drop()
+        end
+    end)
+end)
+
+g.test_invalidate_cache_on_rollback = function(cg)
+    cg.server:exec(function()
+        local fiber = require('fiber')
+
+        local s = box.schema.space.create('test', {engine = 'vinyl'})
+        s:create_index('primary')
+
+        local function read_committed()
+            return box.atomic({txn_isolation = 'read-committed'}, s.select, s)
+        end
+
+        s:insert({10})
+        s:insert({20})
+        s:insert({30})
+        s:insert({40})
+        s:insert({50})
+        box.snapshot()
+
+        box.error.injection.set('ERRINJ_WAL_DELAY', true)
+
+        local fibers = {}
+        table.insert(fibers, fiber.new(s.replace, s, {10, 1}))
+        table.insert(fibers, fiber.new(s.delete, s, {30}))
+        table.insert(fibers, fiber.new(s.replace, s, {60, 1}))
+        for _, f in ipairs(fibers) do
+            f:set_joinable(true)
+        end
+        fiber.yield()
+
+        t.assert_equals(read_committed(), {{10, 1}, {20}, {40}, {50}, {60, 1}})
+
+        box.error.injection.set('ERRINJ_WAL_WRITE', true)
+        box.error.injection.set('ERRINJ_WAL_DELAY', false)
+
+        for _, f in ipairs(fibers) do
+            local ok, err = f:join(5)
+            t.assert_not(ok)
+            t.assert_items_include({
+                box.error.WAL_IO,
+                box.error.CASCADE_ROLLBACK,
+            }, {err.code})
+        end
+
+        t.assert_equals(read_committed(), {{10}, {20}, {30}, {40}, {50}})
+    end)
+end