vinyl: fix compaction crash on disk read error

`vy_slice_stream_next()` clears the return value on failure. This isn't
expected by `vy_write_iterator_merge_step()`, which doesn't update
the source position in the `vy_wirte_iterator::src_heap` in this case.
As a result, an attempt to remove `end_of_key_src` from the heap in
`vy_write_iterator_build_history()` may crash as follows:

```
 # 1  0x572a2ecc21a6 in crash_collect+256
 # 2  0x572a2ecc2be2 in crash_signal_cb+100
 # 3  0x7cfef6645320 in __sigaction+80
 # 4  0x572a2eab16de in tuple_format+16
 # 5  0x572a2eab1a25 in vy_stmt_is_key+24
 # 6  0x572a2eab1be8 in vy_stmt_compare+89
 # 7  0x572a2eab1e37 in vy_entry_compare+74
 # 8  0x572a2eab2913 in heap_less+88
 # 9  0x572a2eab21e3 in vy_source_heap_sift_up+255
 # 10 0x572a2eab20b9 in vy_source_heap_update_node+54
 # 11 0x572a2eab25c1 in vy_source_heap_delete+249
 # 12 0x572a2eab4134 in vy_write_iterator_build_history+1497
 # 13 0x572a2eab4995 in vy_write_iterator_build_read_views+193
 # 14 0x572a2eab4ce6 in vy_write_iterator_next+380
 # 15 0x572a2eadd20b in vy_task_write_run+1132
 # 16 0x572a2eade6cf in vy_task_compaction_execute+124
 # 17 0x572a2eadfa8d in vy_task_f+445
 # 18 0x572a2e9ea143 in fiber_cxx_invoke(int (*)(__va_list_tag*), __va_list_tag*)+34
 # 19 0x572a2eccee7c in fiber_loop+219
 # 20 0x572a2f0aef18 in coro_init+120
```

Normally, a function shouldn't update the return value on failure so
let's fix `vy_slice_stream_next()`.

Closes tarantool#10555

NO_DOC=bug fix

changelogs/unreleased/gh-10555-fix-vinyl-compaction-crash-on-read-error.md

@@ -0,0 +1,4 @@
+## bugfix/vinyl
+
+* Fixed a bug when a compaction task could crash on a disk read error
+  (gh-10555).

src/box/vy_run.c

@@ -2672,11 +2672,12 @@ vy_slice_stream_next(struct vy_stmt_stream *virt_stream, struct vy_entry *ret)
 {
 	assert(virt_stream->iface->next == vy_slice_stream_next);
 	struct vy_slice_stream *stream = (struct vy_slice_stream *)virt_stream;
-	*ret = vy_entry_none();
 
 	/* If the slice is ended, return EOF */
-	if (stream->page_no > stream->slice->last_page_no)
+	if (stream->page_no > stream->slice->last_page_no) {
+		*ret = vy_entry_none();
 		return 0;
+	}
 
 	/* If current page is not already read, read it */
 	if (stream->page == NULL && vy_slice_stream_read_page(stream) != 0)
@@ -2693,6 +2694,7 @@ vy_slice_stream_next(struct vy_stmt_stream *virt_stream, struct vy_entry *ret)
 	    stream->page_no >= stream->slice->last_page_no &&
 	    vy_entry_compare(entry, stream->slice->end, stream->cmp_def) >= 0) {
 		tuple_unref(entry.stmt);
+		*ret = vy_entry_none();
 		return 0;
 	}
 

src/box/vy_stmt.c

@@ -754,6 +754,11 @@ vy_stmt_encode_secondary(struct tuple *value, struct key_def *cmp_def,
 struct tuple *
 vy_stmt_decode(struct xrow_header *xrow, struct tuple_format *format)
 {
+	ERROR_INJECT_COUNTDOWN(ERRINJ_VY_STMT_DECODE_COUNTDOWN, {
+		diag_set(ClientError, ER_INJECTION,
+			 "vinyl statement decode");
+		return NULL;
+	});
 	struct vy_stmt_env *env = format->engine;
 	struct request request;
 	uint64_t key_map = dml_request_key_map(xrow->type);

src/lib/core/errinj.h

@@ -174,6 +174,7 @@ struct errinj {
 	_(ERRINJ_VY_SCHED_TIMEOUT, ERRINJ_DOUBLE, {.dparam = 0}) \
 	_(ERRINJ_VY_SQUASH_TIMEOUT, ERRINJ_DOUBLE, {.dparam = 0}) \
 	_(ERRINJ_VY_STMT_ALLOC_COUNTDOWN, ERRINJ_INT, {.iparam = -1})\
+	_(ERRINJ_VY_STMT_DECODE_COUNTDOWN, ERRINJ_INT, {.iparam = -1})\
 	_(ERRINJ_VY_TASK_COMPLETE, ERRINJ_BOOL, {.bparam = false}) \
 	_(ERRINJ_VY_WRITE_ITERATOR_START_FAIL, ERRINJ_BOOL, {.bparam = false})\
 	_(ERRINJ_WAIT_QUORUM_COUNT, ERRINJ_INT, {.iparam = 0}) \

test/vinyl-luatest/gh_10555_compaction_read_error_test.lua

@@ -0,0 +1,55 @@
+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()
+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_COMPACTION_DELAY', false)
+        box.error.injection.set('ERRINJ_VY_STMT_DECODE_COUNTDOWN', -1)
+    end)
+end)
+
+g.test_compaction_read_error = function(cg)
+    cg.server:exec(function()
+        local s = box.schema.space.create('test', {engine = 'vinyl'})
+        s:create_index('pk')
+
+        -- Delay compaction and create a few runs.
+        box.error.injection.set('ERRINJ_VY_COMPACTION_DELAY', true)
+        for _ = 1, 5 do
+            box.begin()
+            for k = 1, 100 do
+                s:replace({k})
+            end
+            box.commit()
+            box.snapshot()
+        end
+
+        -- Make compaction fail while decoding a random statement.
+        box.error.injection.set('ERRINJ_VY_STMT_DECODE_COUNTDOWN',
+                                math.random(500))
+
+        -- Resume compaction and wait for it to fail.
+        box.stat.reset()
+        t.assert_equals(box.stat.vinyl().scheduler.tasks_failed, 0)
+        s.index.pk:compact()
+        box.error.injection.set('ERRINJ_VY_COMPACTION_DELAY', false)
+        t.helpers.retrying({}, function()
+            t.assert_ge(box.stat.vinyl().scheduler.tasks_failed, 1)
+        end)
+    end)
+end