[PATCH v2] xlog: Remove inprogress files on start

[Ilya Markov <imarkov@tarantool.org>]

When tarantool crashes during writing to vylog, index, run or snapshot,
inprogress files remain. But garbage collector doesn't take into account
these files. So they remain until they are removed manually.

Fix this with adding deletion of such files to functions memtx_engine_end_recovery,
vinyl_engine_end_recovery and vy_run_remove_files.

Add 3 errinj which simulate the crash before renaming inprogress files.

Closes #3406
---
branch: https://github.com/tarantool/tarantool/tree/gh-3406-remove-inprogress-files
issue: https://github.com/tarantool/tarantool/issues/3406

 src/box/memtx_engine.c   |   2 +-
 src/box/vinyl.c          |   1 +
 src/box/vy_log.c         |   4 ++
 src/box/vy_run.c         |  17 +++++
 src/box/xlog.c           |  38 +++++++++++
 src/box/xlog.h           |   6 ++
 src/errinj.h             |   3 +
 test/box/errinj.result   | 168 +++++++++++++++++++++++++++++++++++++++++++++++
 test/box/errinj.test.lua |  63 ++++++++++++++++++
 9 files changed, 301 insertions(+), 1 deletion(-)

diff --git a/src/box/memtx_engine.c b/src/box/memtx_engine.c
index 9a9aff5..3ab60f0 100644
--- a/src/box/memtx_engine.c
+++ b/src/box/memtx_engine.c
@@ -303,7 +303,7 @@ memtx_engine_end_recovery(struct engine *engine)
 		if (space_foreach(memtx_build_secondary_keys, memtx) != 0)
 			return -1;
 	}
-	return 0;
+	return xdir_collect_inprogress(memtx->snap_dir.dirname);
 }

 static struct space *
diff --git a/src/box/vinyl.c b/src/box/vinyl.c
index 552d42b..69646b6 100644
--- a/src/box/vinyl.c
+++ b/src/box/vinyl.c
@@ -2941,6 +2941,7 @@ vinyl_engine_end_recovery(struct engine *engine)
 	case VINYL_FINAL_RECOVERY_LOCAL:
 		if (vy_log_end_recovery() != 0)
 			return -1;
+		xdir_collect_inprogress(e->path);
 		/*
 		 * If the instance is shut down while a dump or
 		 * compaction task is in progress, we'll get an
diff --git a/src/box/vy_log.c b/src/box/vy_log.c
index 51e3753..3519bde 100644
--- a/src/box/vy_log.c
+++ b/src/box/vy_log.c
@@ -848,6 +848,10 @@ vy_log_open(struct xlog *xlog)
 	struct vy_log_record record;
 	vy_log_record_init(&record);
 	record.type = VY_LOG_SNAPSHOT;
+	ERROR_INJECT(ERRINJ_VY_LOG_RENAME, {
+		diag_set(ClientError, ER_INJECTION, "vy_log open");
+		return -1;
+	});
 	if (vy_log_record_encode(&record, &row) < 0 ||
 	    xlog_write_row(xlog, &row) < 0)
 		goto fail_close_xlog;
diff --git a/src/box/vy_run.c b/src/box/vy_run.c
index 980bc4d..1dda93d 100644
--- a/src/box/vy_run.c
+++ b/src/box/vy_run.c
@@ -2021,6 +2021,14 @@ vy_run_write_index(struct vy_run *run, const char *dirpath,
 			goto fail;
 	}

+	ERROR_INJECT(ERRINJ_VY_INDEX_FILE_RENAME, {
+		region_truncate(region, mem_used);
+		xlog_tx_rollback(&index_xlog);
+		xlog_close(&index_xlog, false);
+		diag_set(ClientError, ER_INJECTION,
+			 "vinyl index write");
+		return -1;
+	});
 	if (xlog_tx_commit(&index_xlog) < 0 ||
 	    xlog_flush(&index_xlog) < 0 ||
 	    xlog_rename(&index_xlog) < 0)
@@ -2263,6 +2271,12 @@ vy_run_writer_commit(struct vy_run_writer *writer)
 	run->info.max_key = vy_key_dup(key);
 	if (run->info.max_key == NULL)
 		goto out;
+	ERROR_INJECT(ERRINJ_VY_RUN_RENAME, {
+		region_truncate(&fiber()->gc, region_svp);
+		diag_set(ClientError, ER_INJECTION,
+			 "vinyl writer commit");
+		return -1;
+	});

 	/* Sync data and link the file to the final name. */
 	if (xlog_sync(&writer->data_xlog) < 0 ||
@@ -2439,6 +2453,9 @@ vy_run_remove_files(const char *dir, uint32_t space_id,
 				(long long)run_id); return -1;});
 	int ret = 0;
 	char path[PATH_MAX];
+	vy_index_snprint_path(path, PATH_MAX, dir, space_id, iid);
+	if (xdir_collect_inprogress(path) < 0)
+		return -1;
 	for (int type = 0; type < vy_file_MAX; type++) {
 		vy_run_snprint_path(path, sizeof(path), dir,
 				    space_id, iid, run_id, type);
diff --git a/src/box/xlog.c b/src/box/xlog.c
index 824ad11..b5cc494 100644
--- a/src/box/xlog.c
+++ b/src/box/xlog.c
@@ -621,6 +621,44 @@ xdir_collect_garbage(struct xdir *dir, int64_t signature, bool use_coio)
 	return 0;
 }

+int
+xdir_collect_inprogress(const char *dirname)
+{
+	DIR *dh = opendir(dirname);
+	if (dh == NULL) {
+		if (errno == ENOENT)
+			return 0;
+		diag_set(SystemError, "error reading directory '%s'",
+			 dirname);
+		return -1;
+	}
+
+	struct dirent *dent;
+	char path[PATH_MAX];
+	int total = snprintf(path, sizeof(path), "%s/", dirname);
+	if (total < 0) {
+		say_syserror("error snprintf %s", dirname);
+		diag_set(SystemError, "error snprintf'",
+			 path);
+		return -1;
+	}
+	while ((dent = readdir(dh)) != NULL) {
+		char *ext = strrchr(dent->d_name, '.');
+		if (ext == NULL || strcmp(ext, inprogress_suffix) != 0)
+			continue;
+		strcpy(path + total, dent->d_name);
+		say_info("removing %s", path);
+		int rc = coio_unlink(path);
+		if (rc < 0) {
+			say_syserror("error while removing %s", path);
+			diag_set(SystemError, "failed to unlink file '%s'",
+				 path);
+			return -1;
+		}
+	}
+	return 0;
+}
+
 /* }}} */


diff --git a/src/box/xlog.h b/src/box/xlog.h
index 973910d..4faa763 100644
--- a/src/box/xlog.h
+++ b/src/box/xlog.h
@@ -181,6 +181,12 @@ int
 xdir_collect_garbage(struct xdir *dir, int64_t signature, bool use_coio);

 /**
+ * Remove .inprogress files in specified directory.
+ */
+int
+xdir_collect_inprogress(const char *dirname);
+
+/**
  * Return LSN and vclock (unless @vclock is NULL) of the newest
  * file in a directory or -1 if the directory is empty.
  */
diff --git a/src/errinj.h b/src/errinj.h
index 78514ac..8444479 100644
--- a/src/errinj.h
+++ b/src/errinj.h
@@ -111,6 +111,9 @@ struct errinj {
 	_(ERRINJ_IPROTO_TX_DELAY, ERRINJ_BOOL, {.bparam = false}) \
 	_(ERRINJ_LOG_ROTATE, ERRINJ_BOOL, {.bparam = false}) \
 	_(ERRINJ_SNAP_COMMIT_DELAY, ERRINJ_BOOL, {.bparam = 0}) \
+	_(ERRINJ_VY_LOG_RENAME, ERRINJ_BOOL, {.bparam = false}) \
+	_(ERRINJ_VY_INDEX_FILE_RENAME, ERRINJ_BOOL, {.bparam = false}) \
+	_(ERRINJ_VY_RUN_RENAME, ERRINJ_BOOL, {.bparam = false}) \

 ENUM0(errinj_id, ERRINJ_LIST);
 extern struct errinj errinjs[];
diff --git a/test/box/errinj.result b/test/box/errinj.result
index 6ced172..831c8e3 100644
--- a/test/box/errinj.result
+++ b/test/box/errinj.result
@@ -4,6 +4,12 @@ errinj = box.error.injection
 net_box = require('net.box')
 ---
 ...
+fio = require("fio")
+---
+...
+fiber = require('fiber')
+---
+...
 space = box.schema.space.create('tweedledum')
 ---
 ...
@@ -40,6 +46,8 @@ errinj.info()
     state: false
   ERRINJ_WAL_IO:
     state: false
+  ERRINJ_VY_INDEX_FILE_RENAME:
+    state: false
   ERRINJ_TUPLE_ALLOC:
     state: false
   ERRINJ_VY_READ_PAGE:
@@ -54,8 +62,12 @@ errinj.info()
     state: false
   ERRINJ_VY_RUN_WRITE:
     state: false
+  ERRINJ_VY_RUN_RENAME:
+    state: false
   ERRINJ_VY_LOG_FLUSH_DELAY:
     state: false
+  ERRINJ_VY_LOG_RENAME:
+    state: false
   ERRINJ_SNAP_COMMIT_DELAY:
     state: false
   ERRINJ_RELAY_FINAL_SLEEP:
@@ -693,6 +705,162 @@ errinj.set("ERRINJ_WAL_WRITE", false)
 space:drop()
 ---
 ...
+errinj.set("ERRINJ_VY_LOG_RENAME", true)
+---
+- ok
+...
+s = box.schema.space.create("vinyl_test", {engine='vinyl'})
+---
+...
+_ = s:create_index("primary")
+---
+...
+vinyl_vylog = string.format("%s/*.vylog.inprogress", box.cfg.vinyl_dir)
+---
+...
+#fio.glob(vinyl_vylog) > 0 -- true
+---
+- true
+...
+errinj.set("ERRINJ_SNAP_COMMIT_DELAY", true)
+---
+- ok
+...
+_ = fiber.create(function() box.snapshot() end)
+---
+...
+memtx_snap = string.format("%s/*.snap.inprogress", box.cfg.memtx_dir)
+---
+...
+while #fio.glob(memtx_snap) == 0 do fiber.sleep(0.001) end-- true
+---
+...
+#fio.glob(memtx_snap) > 0 --true
+---
+- true
+...
+test_run:cmd("restart server default")
+errinj = box.error.injection
+---
+...
+net_box = require('net.box')
+---
+...
+fio = require("fio")
+---
+...
+fiber = require('fiber')
+---
+...
+memtx_snap = string.format("%s/*.snap.inprogress", box.cfg.memtx_dir)
+---
+...
+#fio.glob(memtx_snap) == 0 -- true
+---
+- true
+...
+vinyl_vylog = string.format("%s/*.vylog.inprogress", box.cfg.vinyl_dir)
+---
+...
+#fio.glob(vinyl_vylog) == 0 --true
+---
+- true
+...
+s = box.space.vinyl_test
+---
+...
+vinyl_index = string.format("%s/%s/*/*.index.inprogress", box.cfg.vinyl_dir, s.id)
+---
+...
+errinj.set("ERRINJ_VY_INDEX_FILE_RENAME", true)
+---
+- ok
+...
+-- insert big tuples in order to cause compaction without box.snapshot.
+for i = 1, 10000 do s:insert{i, string.rep('a', 10000)} end
+---
+...
+while #fio.glob(vinyl_index) == 0 do fiber.sleep(0.001) end
+---
+...
+vinyl_run = string.format("%s/%s/*/*.run.inprogress", box.cfg.vinyl_dir, s.id)
+---
+...
+errinj.set("ERRINJ_VY_RUN_RENAME", true)
+---
+- ok
+...
+while #fio.glob(vinyl_run) == 0  do fiber.sleep(0.001) end
+---
+...
+#fio.glob(vinyl_run) > 0 -- true
+---
+- true
+...
+#fio.glob(vinyl_index) > 0 -- true
+---
+- true
+...
+test_run:cmd("restart server default")
+errinj = box.error.injection
+---
+...
+net_box = require('net.box')
+---
+...
+fio = require("fio")
+---
+...
+fiber = require('fiber')
+---
+...
+s = box.space.vinyl_test
+---
+...
+memtx_snap = string.format("%s/*.snap.inprogress", box.cfg.memtx_dir)
+---
+...
+vinyl_vylog = string.format("%s/*.vylog.inprogress", box.cfg.vinyl_dir)
+---
+...
+vinyl_run = string.format("%s/%s/*/*.run.inprogress", box.cfg.vinyl_dir, s.id)
+---
+...
+vinyl_index = string.format("%s/%s/*/*.index.inprogress", box.cfg.vinyl_dir, s.id)
+---
+...
+-- no inprogress files should be present in memtx(vinyl)_dir.
+#fio.glob(memtx_snap) == 0 -- true
+---
+- true
+...
+#fio.glob(vinyl_vylog) == 0 -- true
+---
+- true
+...
+#fio.glob(vinyl_index) == 0 -- true
+---
+- true
+...
+#fio.glob(vinyl_run) == 0 -- true
+---
+- true
+...
+box.space.vinyl_test:drop()
+---
+...
+run_dir = fio.pathjoin(box.cfg.vinyl_dir,"*/*/*")
+---
+...
+for _, file in ipairs(fio.glob(run_dir)) do fio.unlink(file) end
+---
+...
+vinyl_dir = fio.pathjoin(box.cfg.vinyl_dir,"*/")
+---
+...
+for _, dir in ipairs(fio.glob(vinyl_dir)) do fio.rmtree(dir) end
+---
+...
 --test space:bsize() in case of memory error
 utils = dofile('utils.lua')
 ---
diff --git a/test/box/errinj.test.lua b/test/box/errinj.test.lua
index 3af1b74..515049a 100644
--- a/test/box/errinj.test.lua
+++ b/test/box/errinj.test.lua
@@ -1,5 +1,7 @@
 errinj = box.error.injection
 net_box = require('net.box')
+fio = require("fio")
+fiber = require('fiber')

 space = box.schema.space.create('tweedledum')
 index = space:create_index('primary', { type = 'hash' })
@@ -203,6 +205,67 @@ box.snapshot()
 errinj.set("ERRINJ_WAL_WRITE", false)
 space:drop()

+errinj.set("ERRINJ_VY_LOG_RENAME", true)
+s = box.schema.space.create("vinyl_test", {engine='vinyl'})
+_ = s:create_index("primary")
+vinyl_vylog = string.format("%s/*.vylog.inprogress", box.cfg.vinyl_dir)
+#fio.glob(vinyl_vylog) > 0 -- true
+
+errinj.set("ERRINJ_SNAP_COMMIT_DELAY", true)
+_ = fiber.create(function() box.snapshot() end)
+memtx_snap = string.format("%s/*.snap.inprogress", box.cfg.memtx_dir)
+while #fio.glob(memtx_snap) == 0 do fiber.sleep(0.001) end-- true
+#fio.glob(memtx_snap) > 0 --true
+
+test_run:cmd("restart server default")
+errinj = box.error.injection
+net_box = require('net.box')
+fio = require("fio")
+fiber = require('fiber')
+
+memtx_snap = string.format("%s/*.snap.inprogress", box.cfg.memtx_dir)
+#fio.glob(memtx_snap) == 0 -- true
+
+vinyl_vylog = string.format("%s/*.vylog.inprogress", box.cfg.vinyl_dir)
+#fio.glob(vinyl_vylog) == 0 --true
+
+s = box.space.vinyl_test
+vinyl_index = string.format("%s/%s/*/*.index.inprogress", box.cfg.vinyl_dir, s.id)
+errinj.set("ERRINJ_VY_INDEX_FILE_RENAME", true)
+-- insert big tuples in order to cause compaction without box.snapshot.
+for i = 1, 10000 do s:insert{i, string.rep('a', 10000)} end
+
+while #fio.glob(vinyl_index) == 0 do fiber.sleep(0.001) end
+
+vinyl_run = string.format("%s/%s/*/*.run.inprogress", box.cfg.vinyl_dir, s.id)
+errinj.set("ERRINJ_VY_RUN_RENAME", true)
+while #fio.glob(vinyl_run) == 0  do fiber.sleep(0.001) end
+#fio.glob(vinyl_run) > 0 -- true
+#fio.glob(vinyl_index) > 0 -- true
+
+test_run:cmd("restart server default")
+errinj = box.error.injection
+net_box = require('net.box')
+fio = require("fio")
+fiber = require('fiber')
+
+s = box.space.vinyl_test
+memtx_snap = string.format("%s/*.snap.inprogress", box.cfg.memtx_dir)
+vinyl_vylog = string.format("%s/*.vylog.inprogress", box.cfg.vinyl_dir)
+vinyl_run = string.format("%s/%s/*/*.run.inprogress", box.cfg.vinyl_dir, s.id)
+vinyl_index = string.format("%s/%s/*/*.index.inprogress", box.cfg.vinyl_dir, s.id)
+
+-- no inprogress files should be present in memtx(vinyl)_dir.
+#fio.glob(memtx_snap) == 0 -- true
+#fio.glob(vinyl_vylog) == 0 -- true
+#fio.glob(vinyl_index) == 0 -- true
+#fio.glob(vinyl_run) == 0 -- true
+
+box.space.vinyl_test:drop()
+run_dir = fio.pathjoin(box.cfg.vinyl_dir,"*/*/*")
+for _, file in ipairs(fio.glob(run_dir)) do fio.unlink(file) end
+vinyl_dir = fio.pathjoin(box.cfg.vinyl_dir,"*/")
+for _, dir in ipairs(fio.glob(vinyl_dir)) do fio.rmtree(dir) end
 --test space:bsize() in case of memory error
 utils = dofile('utils.lua')
 s = box.schema.space.create('space_bsize')
--
2.7.4