box: handle cancelling fiber waiting in WAL queue

Currently if fiber waiting in WAL queue is cancelled it will go on
writing request to WAL breaking queue order. So WAL is corrupted so that
even next Tarantool restart may fail. Unfortunately we cancel fibers on
Tarantool shutdown so the issue may arise just buy terminating Tarantool
instance.

Let's instead fail the cancelled request and all the newer request in
the queue.

Work around is to disable WAL queue by `box.cfg{wal_queue_max_size = 0}`
so that only one write request can be in the queue.

Closes tarantool#11078

NO_DOC=bugfix

Author: nshy

changelogs/unreleased/gh-11078-handle-cancel-of-request-in-wal-queue.md

@@ -0,0 +1,4 @@
+## bugfix/box
+
+* Fixed a bug when cancelling a fiber waiting in WAL queue corrupted the
+  the WAL (gh-11078).

src/box/journal.c

@@ -40,8 +40,7 @@ struct journal *current_journal = NULL;
 struct journal_queue journal_queue = {
 	.max_size = 16 * 1024 * 1024, /* 16 megabytes */
 	.size = 0,
-	.waiters = RLIST_HEAD_INITIALIZER(journal_queue.waiters),
-	.waiter_count = 0,
+	.requests = STAILQ_INITIALIZER(journal_queue.requests),
 };
 
 void
@@ -94,36 +93,73 @@ journal_entry_fiber_wakeup_cb(struct journal_entry *entry)
 void
 journal_queue_wakeup(void)
 {
-	struct rlist *list = &journal_queue.waiters;
-	if (!rlist_empty(list) && !journal_queue_is_full())
-		fiber_wakeup(rlist_first_entry(list, struct fiber, state));
+	if (!stailq_empty(&journal_queue.requests) &&
+	    !journal_queue_is_full()) {
+		struct journal_entry *req =
+				stailq_first_entry(&journal_queue.requests,
+						   typeof(*req), fifo);
+		fiber_wakeup(req->fiber);
+	}
 }
 
-void
-journal_queue_wait(void)
+int
+journal_queue_wait(struct journal_entry *entry)
 {
-	if (!journal_queue_is_full() && !journal_queue_has_waiters())
-		return;
-	++journal_queue.waiter_count;
-	rlist_add_tail_entry(&journal_queue.waiters, fiber(), state);
-	/*
-	 * Is woken up when this position in the queue should go into the next
-	 * journal batch.
-	 */
+	if (!journal_queue_is_full() &&
+	    stailq_empty(&journal_queue.requests))
+		return 0;
+	int rc = -1;
+	struct journal_entry *prev_entry =
+			stailq_last_entry(&journal_queue.requests,
+					  typeof(*prev_entry), fifo);
+	stailq_add_tail_entry(&journal_queue.requests, entry, fifo);
+	assert(entry->fiber == NULL);
+	entry->fiber = fiber();
 	fiber_yield();
-	--journal_queue.waiter_count;
-	journal_queue_wakeup();
+	if (entry->is_complete) {
+		/* Already rolled back on cascade rollback. */
+		diag_set_journal_res(entry->res);
+	} else if (fiber_is_cancelled()) {
+		struct stailq rollback;
+		stailq_cut_tail(&journal_queue.requests, &prev_entry->fifo,
+				&rollback);
+		/* Pop this request. */
+		VERIFY(stailq_shift_entry(&rollback,
+					  typeof(*entry), fifo) == entry);
+		stailq_reverse(&rollback);
+		/* Cascade rollback of newer requests. */
+		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);
+		}
+		/* Rollback this request. */
+		entry->res = JOURNAL_ENTRY_ERR_CANCELLED;
+		entry->is_complete = true;
+		entry->write_async_cb(entry);
+		diag_set(FiberIsCancelled);
+	} else {
+		/* There is a space in queue to handle this request. */
+		VERIFY(stailq_shift_entry(&journal_queue.requests,
+					  typeof(*entry), fifo) == entry);
+		journal_queue_wakeup();
+		rc = 0;
+	}
+	entry->fiber = NULL;
+	return rc;
 }
 
 void
 journal_queue_flush(void)
 {
-	if (!journal_queue_has_waiters())
+	if (stailq_empty(&journal_queue.requests))
 		return;
-	struct rlist *list = &journal_queue.waiters;
-	while (!rlist_empty(list))
-		fiber_wakeup(rlist_first_entry(list, struct fiber, state));
-	journal_queue_wait();
+	struct journal_entry *req;
+	stailq_foreach_entry(req, &journal_queue.requests, fifo)
+		fiber_wakeup(req->fiber);
+	/* Schedule after all fibers waiting in journal queue. */
+	fiber_sleep(0);
 }
 
 int

src/box/journal.h

@@ -54,6 +54,8 @@ enum {
 	 * entry.
 	 */
 	JOURNAL_ENTRY_ERR_CASCADE = -3,
+	/** Rollback due to fiber waiting for WAL is cancelled. */
+	JOURNAL_ENTRY_ERR_CANCELLED = -4,
 	/**
 	 * Anchor for the structs built on top of journal entry so as they
 	 * could introduce their own unique errors. Set to a big value in
@@ -106,6 +108,10 @@ struct journal_entry {
 	 * journal entry is completed.
 	 */
 	bool is_complete;
+	/**
+	 * Fiber which put the request in journal queue.
+	 */
+	struct fiber *fiber;
 	/**
 	 * The number of rows in the request.
 	 */
@@ -134,6 +140,7 @@ journal_entry_create(struct journal_entry *entry, size_t n_rows,
 	entry->res		= JOURNAL_ENTRY_ERR_UNKNOWN;
 	entry->flags		= 0;
 	entry->is_complete = false;
+	entry->fiber = NULL;
 }
 
 /**
@@ -158,14 +165,8 @@ struct journal_queue {
 	int64_t max_size;
 	/** Current approximate size of journal queue. */
 	int64_t size;
-	/**
-	 * The fibers waiting for some space to free in journal queue.
-	 * Once some space is freed they will be waken up in the same order they
-	 * entered the queue.
-	 */
-	struct rlist waiters;
-	/** How many waiters there are in a queue. */
-	int waiter_count;
+	/** The requests waiting in journal queue. */
+	struct stailq requests;
 };
 
 /** A single queue for all journal instances. */
@@ -186,16 +187,6 @@ struct journal {
 void
 journal_queue_wakeup(void);
 
-/**
- * Check whether anyone is waiting for the journal queue to empty. If there are
- * other waiters we must go after them to preserve write order.
- */
-static inline bool
-journal_queue_has_waiters(void)
-{
-	return journal_queue.waiter_count != 0;
-}
-
 /**
  * Check whether any of the queue size limits is reached.
  * If the queue is full, we must wait for some of the entries to be written
@@ -208,10 +199,10 @@ journal_queue_is_full(void)
 }
 
 /** Yield until there's some space in the journal queue. */
-void
-journal_queue_wait(void);
+int
+journal_queue_wait(struct journal_entry *entry);
 
-/** Empty the queue by waking everyone in it up and put self to queue tail. */
+/** Flush journal queue. Next wal_sync() will sync flushed requests. */
 void
 journal_queue_flush(void);
 
@@ -271,7 +262,8 @@ journal_write_row(struct xrow_header *row);
 static inline int
 journal_write_submit(struct journal_entry *entry)
 {
-	journal_queue_wait();
+	if (journal_queue_wait(entry) != 0)
+		return -1;
 	/*
 	 * We cannot account entry after write. If journal is synchronous
 	 * the journal_queue_on_complete() is called in write_async().

src/box/txn.c

@@ -1179,10 +1179,8 @@ txn_commit_impl(struct txn *txn, enum txn_commit_wait_mode wait_mode)
 		goto rollback_abort;
 	}
 	fiber_set_txn(fiber(), NULL);
-	if (journal_write_submit(req) != 0) {
-		fiber_set_txn(fiber(), txn);
+	if (journal_write_submit(req) != 0)
 		goto rollback_io;
-	}
 	if (wait_mode != TXN_COMMIT_WAIT_MODE_COMPLETE) {
 		if (txn_has_flag(txn, TXN_IS_DONE))
 			goto finish_done;

src/box/txn.h

@@ -144,6 +144,7 @@ enum {
 	TXN_SIGNATURE_UNKNOWN = JOURNAL_ENTRY_ERR_UNKNOWN,
 	TXN_SIGNATURE_IO = JOURNAL_ENTRY_ERR_IO,
 	TXN_SIGNATURE_CASCADE = JOURNAL_ENTRY_ERR_CASCADE,
+	TXN_SIGNATURE_CANCELLED = JOURNAL_ENTRY_ERR_CANCELLED,
 	/**
 	 * The default signature value for failed transactions.
 	 * Indicates either write failure or any other failure

test/box-luatest/gh_11078_wal_queue_fiber_cancel_test.lua

@@ -0,0 +1,80 @@
+local server = require('luatest.server')
+local t = require('luatest')
+
+local variants = {}
+for i = 1, 5 do
+    table.insert(variants, {pos = i})
+end
+local g = t.group('wal', variants)
+
+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()
+        box.error.injection.set('ERRINJ_WAL_DELAY', false)
+        box.space.test:drop()
+    end)
+end)
+
+g.test_wal_queue_fiber_cancel = function(cg)
+    cg.server:exec(function(pos)
+        local fiber = require('fiber')
+        local s = box.schema.create_space('test')
+        s:create_index('pk')
+        box.cfg{wal_queue_max_size = 100}
+        box.error.injection.set('ERRINJ_WAL_DELAY', true)
+        fiber.create(function()
+            box.begin()
+            s:insert({100, string.rep('a', 1000)})
+            s:insert({0})
+            box.commit()
+        end)
+        local fibers = {}
+        for i = 1, 5 do
+            local f = fiber.new(function()
+                -- Crafted to check that WAL records are in correct order
+                -- on recovery after restart.
+                box.begin()
+                s:delete({i - 1})
+                s:insert({i})
+                box.commit()
+            end)
+            f:set_joinable(true)
+            table.insert(fibers, f)
+            fiber.yield()
+        end
+        fibers[pos]:cancel()
+        box.error.injection.set('ERRINJ_WAL_DELAY', false)
+        for i = 1, pos - 1 do
+            t.assert_equals({fibers[i]:join()}, {true})
+        end
+        local ok, err = fibers[pos]:join()
+        t.assert_not(ok)
+        t.assert_covers(err:unpack(), {
+            type = 'FiberIsCancelled',
+        })
+        for i = pos + 1, 5 do
+            local ok, err = fibers[i]:join()
+            t.assert_not(ok)
+            t.assert_covers(err:unpack(), {
+                type = 'ClientError',
+                code = box.error.CASCADE_ROLLBACK,
+                message = "WAL has a rollback in progress",
+            })
+        end
+        t.assert_equals(s:select({100}, {iterator = 'lt'}), {{pos - 1}})
+    end, {cg.params.pos})
+    cg.server:restart()
+    cg.server:exec(function(pos)
+        local s = box.space.test
+        t.assert_equals(s:select({100}, {iterator = 'lt'}), {{pos - 1}})
+    end, {cg.params.pos})
+end