wal: use existing logs sizes after restart

During the `checkpoint_wal_threshold` exceedance checks tarantool used
to ignore the sizes of files created prior to the server restart which
resulted in cases when the server didn't create any checkpoints and
generate a huge amount of logs which were replayed each server restart.
Now the `wal_dir` is scanned for proper `writer->checkpoint_wal_size`
calculation after the recovery process.

Fixes tarantool#9811
NO_DOC=bugfix

Author: LevKats

changelogs/unreleased/gh-9811-checkpoint-wal-threshold-after-restart.md

@@ -0,0 +1,5 @@
+## bugfix/box
+
+* Fixed a bug when the sizes of *.xlog files created before the server restart
+  were not taken into account during the `checkpoint_wal_threshold` exceedance
+  checks (gh-9811).

src/box/box.cc

@@ -6348,9 +6348,14 @@ box_storage_init(void)
 		cfg_geti64("wal_max_size"));
 	enum wal_mode wal_mode = box_check_wal_mode(cfg_gets("wal_mode"));
 	double wal_retention_period = box_check_wal_retention_period_xc();
+	struct vclock *checkpoint_vclock = NULL;
+	struct gc_checkpoint *last_checkpoint = gc_last_checkpoint();
+	if (last_checkpoint != NULL)
+		checkpoint_vclock = &last_checkpoint->vclock;
 	if (wal_init(wal_mode, cfg_gets("wal_dir"), wal_max_size,
 		     wal_retention_period, &INSTANCE_UUID,
-		     &instance_vclock_storage, on_wal_garbage_collection,
+		     &instance_vclock_storage, checkpoint_vclock,
+		     on_wal_garbage_collection,
 		     on_wal_checkpoint_threshold) != 0) {
 		diag_raise();
 	}

src/box/wal.c

@@ -418,6 +418,7 @@ wal_writer_create(struct wal_writer *writer, enum wal_mode wal_mode,
 		  double wal_retention_period,
 		  const struct tt_uuid *instance_uuid,
 		  struct vclock *instance_vclock,
+		  struct vclock *checkpoint_vclock,
 		  wal_on_garbage_collection_f on_garbage_collection,
 		  wal_on_checkpoint_threshold_f on_checkpoint_threshold)
 {
@@ -450,6 +451,8 @@ wal_writer_create(struct wal_writer *writer, enum wal_mode wal_mode,
 
 	vclock_create(&writer->vclock);
 	vclock_create(&writer->checkpoint_vclock);
+	if (checkpoint_vclock != NULL)
+		vclock_copy(&writer->checkpoint_vclock, checkpoint_vclock);
 	rlist_create(&writer->watchers);
 
 	writer->on_garbage_collection = on_garbage_collection;
@@ -478,11 +481,30 @@ wal_open_f(struct cbus_call_msg *msg)
 	const char *path = xdir_format_filename(&writer->wal_dir,
 				vclock_sum(&writer->vclock), NONE);
 	assert(!xlog_is_open(&writer->current_wal));
+
+	uint64_t sum_already_on_disk = 0;
+	for (struct vclock *vclock = vclockset_first(&writer->wal_dir.index);
+	     vclock != NULL;
+	     vclock = vclockset_next(&writer->wal_dir.index, vclock)) {
+		if (vclock_compare(&writer->checkpoint_vclock, vclock) > 0)
+			continue;
+		const char *name = xdir_format_filename(&writer->wal_dir,
+							vclock_sum(vclock),
+							NONE);
+		struct stat attr;
+		if (stat(name, &attr) != 0)
+			say_warn("failed to get file size %s, consider it"
+				 " equal to zero", name);
+		else
+			sum_already_on_disk += (uint64_t)attr.st_size;
+	}
+	writer->checkpoint_wal_size = sum_already_on_disk;
 	return xlog_open(&writer->current_wal, path, &writer->wal_dir.opts);
 }
 
 /**
- * Try to open the current WAL file for appending if it exists.
+ * Try to open the current WAL file for appending if it exists
+ * and scan writer->wal_dir to set writer->checkpoint_wal_size.
  */
 static int
 wal_open(struct wal_writer *writer)
@@ -544,14 +566,16 @@ wal_init(enum wal_mode wal_mode, const char *wal_dirname,
 	 int64_t wal_max_size, double wal_retention_period,
 	 const struct tt_uuid *instance_uuid,
 	 struct vclock *instance_vclock,
+	 struct vclock *checkpoint_vclock,
 	 wal_on_garbage_collection_f on_garbage_collection,
 	 wal_on_checkpoint_threshold_f on_checkpoint_threshold)
 {
 	/* Initialize the state. */
 	struct wal_writer *writer = &wal_writer_singleton;
 	wal_writer_create(writer, wal_mode, wal_dirname, wal_max_size,
 			  wal_retention_period, instance_uuid, instance_vclock,
-			  on_garbage_collection, on_checkpoint_threshold);
+			  checkpoint_vclock, on_garbage_collection,
+			  on_checkpoint_threshold);
 
 	/* Start WAL thread. */
 	if (cord_costart(&writer->cord, "wal", wal_writer_f, NULL) != 0)

src/box/wal.h

@@ -110,6 +110,7 @@ wal_init(enum wal_mode wal_mode, const char *wal_dirname,
 	 int64_t wal_max_size, double wal_retention_period,
 	 const struct tt_uuid *instance_uuid,
 	 struct vclock *instance_vclock,
+	 struct vclock *checkpoint_vclock,
 	 wal_on_garbage_collection_f on_garbage_collection,
 	 wal_on_checkpoint_threshold_f on_checkpoint_threshold);
 

test/box-luatest/gh_9811_checkpoint_wal_threshold_after_restart_test.lua

@@ -0,0 +1,84 @@
+local t = require('luatest')
+local server = require('luatest.server')
+
+-- 1. Many small xlogs. Let's check that threshold check indeed use
+--    the sum of xlog sizes.
+-- 2. One big xlog. Let's check that xlog right after the snapshot is
+--    taken into account.
+local group_config = {{wal_max_size = 50000}, {wal_max_size = 50000 * 10}}
+local g = t.group("wal_max_size", group_config)
+
+g.before_each(function(cg)
+    cg.server = server:new({box_cfg = {
+        wal_max_size = cg.params.wal_max_size,
+        checkpoint_wal_threshold = 10 * 50000,
+        checkpoint_count = 1e8,
+        checkpoint_interval = 0,
+    }})
+    cg.server:start()
+    cg.server:exec(function()
+        local s = box.schema.space.create('test')
+        s:create_index('pk', {sequence = true})
+        -- Let's delete all existing xlogs so ones that were produced during
+        -- server start and index creation don't interfere with our test and
+        -- we don't have to take the sizes of that xlogs into account.
+        -- 1. Reconfigure so each new checkpoint results in GC trigger and
+        --    deletion of all logs before the checkpoint.
+        box.cfg{checkpoint_count = 1}
+        -- 2. Delete all unintended xlogs.
+        box.snapshot()
+        -- 3. Effectively disable GC again.
+        box.cfg{checkpoint_count = 1e8}
+    end)
+end)
+
+g.test_box_checkpoint_wal_threshold_after_restart = function(cg)
+    -- 1. Let's create some xlogs followed by a snapshot.
+    cg.server:exec(function()
+        local fiber = require("fiber")
+        local s = box.space.test
+        local checkpoints_before = #box.info.gc().checkpoints
+        for _ = 1, 7000 do
+            s:insert({box.NULL})
+        end
+        -- Wait for possible (but unwanted) checkpoint finish.
+        fiber.sleep(1)
+        t.assert_equals(#box.info.gc().checkpoints, checkpoints_before,
+                        "Wrong batch size. Change the number of tuples")
+        box.snapshot()
+    end)
+    cg.server:restart()
+    -- 2. There are xlogs created before the snapshot. Check that they are
+    --    ignored when calculating the threshold.
+    cg.server:exec(function()
+        local fiber = require("fiber")
+        local s = box.space.test
+        local checkpoints_before = #box.info.gc().checkpoints
+        for _ = 1, 7000 do
+            s:insert({box.NULL})
+        end
+        -- Result must be the same -- no new checkpoint.
+        fiber.sleep(1)
+        t.assert_equals(#box.info.gc().checkpoints, checkpoints_before,
+                        "old xlogs may be used")
+    end)
+    cg.server:restart()
+    -- 3. We expect a checkpoint after the batch of tuples below since again
+    --    we have some xlogs before the restart but this time there were no
+    --    new snapshots.
+    cg.server:exec(function()
+        local fiber = require("fiber")
+        local checkpoints_before = #box.info.gc().checkpoints
+        local s = box.space.test
+        -- A checkpoint must be created after that batch.
+        for _ = 1, 7000 do
+            s:insert({box.NULL})
+        end
+        fiber.sleep(1)
+        t.assert_gt(#box.info.gc().checkpoints, checkpoints_before)
+    end)
+end
+
+g.after_each(function(cg)
+    cg.server:drop()
+end)