box: rollback txns in WAL queue before in-flight tnxs

In case of WAL error we should rollback txns in WAL queue before any
in-flight txns (already submitted to WAL thread).

Work around is to disable WAL queue by `box.cfg{wal_queue_max_size = 0}`
so that no request can be queued.

Closes tarantool#11179

NO_DOC=bugfix

Author: nshy

changelogs/unreleased/gh-11179-wal-queue-rollback.md

@@ -0,0 +1,4 @@
+## bugfix/box
+
+* Fixed a bug when transactions in the WAL queue were not rolled back on WAL
+  error (gh-11179).

src/box/journal.c

@@ -162,6 +162,21 @@ journal_queue_flush(void)
 	fiber_sleep(0);
 }
 
+void
+journal_queue_rollback(void)
+{
+	struct stailq rollback;
+	stailq_create(&rollback);
+	stailq_concat(&rollback, &journal_queue.requests);
+	stailq_reverse(&rollback);
+	struct journal_entry *req;
+	stailq_foreach_entry(req, &rollback, fifo) {
+		req->res = JOURNAL_ENTRY_ERR_CASCADE;
+		req->is_complete = true;
+		req->write_async_cb(req);
+	}
+}
+
 int
 journal_write_row(struct xrow_header *row)
 {

src/box/journal.h

@@ -229,6 +229,10 @@ journal_queue_on_complete(const struct journal_entry *entry)
 	assert(journal_queue.size >= 0);
 }
 
+/** Rollback all txns waiting in queue. */
+void
+journal_queue_rollback(void);
+
 /**
  * Complete asynchronous write.
  */
@@ -271,6 +275,7 @@ journal_write_submit(struct journal_entry *entry)
 	journal_queue_on_append(entry);
 	if (current_journal->write_async(current_journal, entry) != 0) {
 		journal_queue_on_complete(entry);
+		journal_queue_rollback();
 		return -1;
 	}
 	return 0;

src/box/wal.c

@@ -337,6 +337,7 @@ tx_complete_rollback(void)
 	if (stailq_last_entry(&writer->rollback, struct journal_entry,
 			      fifo) != writer->last_entry)
 		return;
+	journal_queue_rollback();
 	stailq_reverse(&writer->rollback);
 	tx_schedule_queue(&writer->rollback);
 	/* TX-thread can try sending transactions to WAL again. */

test/box-luatest/gh_11179_wal_queue_rollback_test.lua

@@ -0,0 +1,142 @@
+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_test('test_wal_queue_rollback_in_flight', function(cg)
+    cg.server:exec(function()
+        box.error.injection.set('ERRINJ_WAL_DELAY', false)
+        box.error.injection.set('ERRINJ_WAL_WRITE', false)
+        box.space.test:drop()
+    end)
+end)
+
+g.test_wal_queue_rollback_in_flight = function(cg)
+    cg.server:exec(function()
+        local fiber = require('fiber')
+        local s = box.schema.create_space('test')
+        s:create_index('pk')
+        box.cfg{wal_queue_max_size = 100}
+        s:insert({1})
+        box.error.injection.set('ERRINJ_WAL_DELAY', true)
+        -- In case txn in WAL queue (f2) is not rollbacked we get duplicate
+        -- error on rollback of in-flight txn (f1) and as a result failed
+        -- assertion or panic.
+        local f1 = fiber.new(function()
+            box.begin()
+            s:delete({1})
+            s:insert({100, string.rep('a', 1000)})
+            box.commit()
+        end)
+        f1:set_joinable(true)
+        fiber.yield()
+        local f2 = fiber.new(function()
+            s:insert({1})
+        end)
+        f2:set_joinable(true)
+        fiber.yield()
+        box.error.injection.set('ERRINJ_WAL_WRITE', true)
+        box.error.injection.set('ERRINJ_WAL_DELAY', false)
+        local ok, err = f1:join()
+        t.assert_not(ok)
+        t.assert_covers(err:unpack(), {
+            type = 'ClientError',
+            code = box.error.WAL_IO,
+            message = 'Failed to write to disk',
+        })
+        local ok, err = f2:join()
+        t.assert_not(ok)
+        t.assert_covers(err:unpack(), {
+            type = 'ClientError',
+            code = box.error.CASCADE_ROLLBACK,
+            message = 'WAL has a rollback in progress',
+        })
+        t.assert_equals(s:select(), {{1}})
+    end)
+end
+
+g.after_test('test_wal_queue_rollback_cascade', function(cg)
+    cg.server:exec(function()
+        box.error.injection.set('ERRINJ_WAL_DELAY', false)
+        box.error.injection.set('ERRINJ_WAL_IO', false)
+        box.space.test:drop()
+    end)
+end)
+
+--
+-- Here we test a different situation. Part of in-flight requests can be
+-- successfully written and part of requests are not. Imagine also in-flight
+-- requests are split into 2 batches. The first has both successful and not
+-- successful and the second has only unsuccessful requests. So when first
+-- batch is returned to TX thread we don't proceed with rollback yet
+-- waiting for the second batch. But we complete the successful part of
+-- the batch and thus wakeup journal queue. Woken up fiber will try
+-- to submit new request to WAL but fail as cascade rollback is in
+-- progress. The failed request from journal will be rolled back and
+-- this is not correct. First we should rollback newer requests form the
+-- journal queue.
+--
+-- Why do we need two batches here? With only one batch we first rollback
+-- failed requests from batch and thus rollback journal queue.
+--
+-- Note also that this situation it hard to reproduce directly. Thus it
+-- modelled here by ERRINJ_WAL_IO injection.
+--
+g.test_wal_queue_rollback_cascade = function(cg)
+    cg.server:exec(function()
+        local fiber = require('fiber')
+        local s = box.schema.create_space('test')
+        s:create_index('pk')
+        s:insert({1})
+        box.cfg{wal_queue_max_size = 100}
+        box.error.injection.set('ERRINJ_WAL_DELAY', true)
+        local f1 = fiber.new(function()
+            box.begin()
+            s:insert({100, string.rep('a', 1000)})
+            box.commit()
+        end)
+        f1:set_joinable(true)
+        fiber.yield()
+        -- In case txn in WAL queue (f3) is not rollbacked we get duplicate
+        -- error on rollback of in-flight txn (f2) and as a result failed
+        -- assertion or panic.
+        local f2 = fiber.new(function()
+            s:delete({1})
+        end)
+        f2:set_joinable(true)
+        fiber.yield()
+        local f3 = fiber.new(function()
+            s:insert({1})
+        end)
+        f3:set_joinable(true)
+        fiber.yield()
+        box.error.injection.set('ERRINJ_WAL_IO', true)
+        box.error.injection.set('ERRINJ_WAL_DELAY', false)
+        t.assert_equals({f1:join()}, {true})
+        local ok, err = f2:join()
+        t.assert_not(ok)
+        t.assert_covers(err:unpack(), {
+            type = 'ClientError',
+            code = box.error.WAL_IO,
+            message = 'Failed to write to disk',
+        })
+        local ok, err = f3:join()
+        t.assert_not(ok)
+        t.assert_covers(err:unpack(), {
+            type = 'ClientError',
+            code = box.error.CASCADE_ROLLBACK,
+            message = 'WAL has a rollback in progress',
+        })
+        t.assert_equals(s:select({100}, {iterator = 'lt'}), {{1}})
+    end)
+end