[tarantool-patches] [xlog 1/1] 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 function to box_cfg which traverses memtx_dir and
vinyl_dir and removes inprogress files.

* Add 4 errinj which simulate the crash before renaming inprogress
  files.
* Change signature of engine_commit_checkpoint function in order to
support error injection in snapshot commit.

Closes #3406
---
branch: gh-3406-remove-inprogress-files
 src/box/box.cc           |  20 +++++-
 src/box/engine.c         |   3 +-
 src/box/engine.h         |   2 +-
 src/box/memtx_engine.c   |  21 +++++-
 src/box/sysview_engine.c |   3 +-
 src/box/vinyl.c          |   3 +-
 src/box/vy_log.c         |   4 ++
 src/box/vy_run.c         |  14 ++++
 src/box/xlog.c           |  73 +++++++++++++++++++-
 src/box/xlog.h           |  13 ++++
 src/errinj.h             |   4 ++
 test/box/errinj.result   | 171 ++++++++++++++++++++++++++++++++++++++++++++++-
 test/box/errinj.test.lua |  59 ++++++++++++++++
 13 files changed, 378 insertions(+), 12 deletions(-)

diff --git a/src/box/box.cc b/src/box/box.cc
index e3eb273..1e75b15 100644
--- a/src/box/box.cc
+++ b/src/box/box.cc
@@ -1683,6 +1683,24 @@ tx_prio_cb(struct ev_loop *loop, ev_watcher *watcher, int events)
 	cbus_process(endpoint);
 }
 
+/**
+ * Removes .inprogress files in memtx_dir, vinyl_dir and vinyl_dir subdirs.
+ */
+static void
+collect_vinyl_subdirectories_inprogress()
+{
+	/* Remove inprogress files in directories. */
+	if (xdir_collect_inprogress(cfg_gets("memtx_dir")) < 0 ||
+	    (strcmp(cfg_gets("memtx_dir"),
+		    cfg_gets("vinyl_dir")) != 0 &&
+	     xdir_collect_inprogress(
+		     cfg_gets("vinyl_dir")) < 0))
+		diag_raise();
+	if (space_foreach(collect_vinyl_inprogress,
+			  (void *) cfg_gets("vinyl_dir")) < 0)
+		diag_raise();
+}
+
 void
 box_init(void)
 {
@@ -1810,7 +1828,6 @@ box_cfg_xc(void)
 		 */
 		memtx_engine_recover_snapshot_xc(memtx,
 				&last_checkpoint_vclock);
-
 		engine_begin_final_recovery_xc();
 		recovery_follow_local(recovery, &wal_stream.base, "hot_standby",
 				      cfg_getd("wal_dir_rescan_delay"));
@@ -1867,6 +1884,7 @@ box_cfg_xc(void)
 
 		/* Wait for the cluster to start up */
 		box_sync_replication(replication_connect_timeout, false);
+		collect_vinyl_subdirectories_inprogress();
 	} else {
 		if (!tt_uuid_is_nil(&instance_uuid))
 			INSTANCE_UUID = instance_uuid;
diff --git a/src/box/engine.c b/src/box/engine.c
index e4ae156..8a0e885 100644
--- a/src/box/engine.c
+++ b/src/box/engine.c
@@ -134,7 +134,8 @@ engine_commit_checkpoint(struct vclock *vclock)
 			return -1;
 	}
 	engine_foreach(engine) {
-		engine->vtab->commit_checkpoint(engine, vclock);
+		if (engine->vtab->commit_checkpoint(engine, vclock) < 0)
+			return -1;
 	}
 	return 0;
 }
diff --git a/src/box/engine.h b/src/box/engine.h
index 5f8b589..80b3306 100644
--- a/src/box/engine.h
+++ b/src/box/engine.h
@@ -149,7 +149,7 @@ struct engine_vtab {
 	 * All engines prepared their checkpoints,
 	 * fix up the changes.
 	 */
-	void (*commit_checkpoint)(struct engine *, struct vclock *);
+	int (*commit_checkpoint)(struct engine *, struct vclock *);
 	/**
 	 * An error in one of the engines, abort checkpoint.
 	 */
diff --git a/src/box/memtx_engine.c b/src/box/memtx_engine.c
index 9a9aff5..bb350a9 100644
--- a/src/box/memtx_engine.c
+++ b/src/box/memtx_engine.c
@@ -703,7 +703,7 @@ memtx_engine_wait_checkpoint(struct engine *engine, struct vclock *vclock)
 	return result;
 }
 
-static void
+static int
 memtx_engine_commit_checkpoint(struct engine *engine, struct vclock *vclock)
 {
 	(void) vclock;
@@ -732,9 +732,16 @@ memtx_engine_commit_checkpoint(struct engine *engine, struct vclock *vclock)
 				fiber_sleep(0.001);
 		}
 #endif
+		ERROR_INJECT(ERRINJ_SNAP_MEMTX, {
+			diag_set(ClientError, ER_INJECTION,
+				 "memtx snapshot commit");
+			return -1;
+		});
 		int rc = coio_rename(from, to);
-		if (rc != 0)
-			panic("can't rename .snap.inprogress");
+		if (rc != 0) {
+			diag_set(SystemError, "can't rename .snap.inprogress");
+			return -1;
+		}
 	}
 
 	struct vclock last;
@@ -748,6 +755,7 @@ memtx_engine_commit_checkpoint(struct engine *engine, struct vclock *vclock)
 
 	checkpoint_destroy(memtx->checkpoint);
 	memtx->checkpoint = NULL;
+	return 0;
 }
 
 static void
@@ -767,6 +775,13 @@ memtx_engine_abort_checkpoint(struct engine *engine)
 
 	memtx_tuple_end_snapshot();
 
+	ERROR_INJECT(ERRINJ_SNAP_MEMTX, {
+		/* Don't remove inprogress files. */
+		checkpoint_destroy(memtx->checkpoint);
+		memtx->checkpoint = NULL;
+		return;
+	});
+
 	/** Remove garbage .inprogress file. */
 	char *filename =
 		xdir_format_filename(&memtx->checkpoint->dir,
diff --git a/src/box/sysview_engine.c b/src/box/sysview_engine.c
index b478e78..b63bfc0 100644
--- a/src/box/sysview_engine.c
+++ b/src/box/sysview_engine.c
@@ -335,11 +335,12 @@ sysview_engine_wait_checkpoint(struct engine *engine, struct vclock *vclock)
 	return 0;
 }
 
-static void
+static int
 sysview_engine_commit_checkpoint(struct engine *engine, struct vclock *vclock)
 {
 	(void)engine;
 	(void)vclock;
+	return 0;
 }
 
 static void
diff --git a/src/box/vinyl.c b/src/box/vinyl.c
index 552d42b..ee4dbd8 100644
--- a/src/box/vinyl.c
+++ b/src/box/vinyl.c
@@ -2861,13 +2861,14 @@ vinyl_engine_wait_checkpoint(struct engine *engine, struct vclock *vclock)
 	return 0;
 }
 
-static void
+static int
 vinyl_engine_commit_checkpoint(struct engine *engine, struct vclock *vclock)
 {
 	(void)vclock;
 	struct vy_env *env = vy_env(engine);
 	assert(env->status == VINYL_ONLINE);
 	vy_scheduler_end_checkpoint(&env->scheduler);
+	return 0;
 }
 
 static void
diff --git a/src/box/vy_log.c b/src/box/vy_log.c
index 51e3753..cea3080 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_OPEN, {
+		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..6b41620 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_CREATE, {
+		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_COMMIT, {
+		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 ||
diff --git a/src/box/xlog.c b/src/box/xlog.c
index 824ad11..1750d52 100644
--- a/src/box/xlog.c
+++ b/src/box/xlog.c
@@ -46,6 +46,7 @@
 #include "xrow.h"
 #include "iproto_constants.h"
 #include "errinj.h"
+#include "space.h"
 
 /*
  * marker is MsgPack fixext2
@@ -621,6 +622,77 @@ 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);
+		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;
+}
+
+int
+collect_vinyl_inprogress(struct space *space, void *param)
+{
+	if (space->def->engine_name[0] != 'v')
+		return 0;
+	const char * vinyl_dir = (const char *) param;
+	static char path[PATH_MAX];
+	int total = snprintf(path, sizeof(path), "%s/%i/", vinyl_dir,
+			     space->def->id);
+	if (total < 0) {
+		say_syserror("error snprintf %s", vinyl_dir);
+		diag_set(SystemError, "error snprintf'",
+			 path);
+		return -1;
+	}
+
+	DIR *dh = opendir(path);
+	if (dh == NULL) {
+		if (errno == ENOENT)
+			return 0;
+		diag_set(SystemError, "error reading directory '%s'",
+			 path);
+		return -1;
+	}
+	struct dirent *dent;
+	while ((dent = readdir(dh)) != NULL) {
+		/* Add iid to path */
+		strcpy(path + total, dent->d_name);
+		if (xdir_collect_inprogress(path) < 0)
+			return -1;
+	}
+	return 0;
+}
+
 /* }}} */
 
 
@@ -636,7 +708,6 @@ xlog_rename(struct xlog *l)
 	assert(l->is_inprogress);
 	assert(suffix);
 	assert(strcmp(suffix, inprogress_suffix) == 0);
-
 	/* Create a new filename without '.inprogress' suffix. */
 	memcpy(new_filename, filename, suffix - filename);
 	new_filename[suffix - filename] = '\0';
diff --git a/src/box/xlog.h b/src/box/xlog.h
index 973910d..cb3c8b0 100644
--- a/src/box/xlog.h
+++ b/src/box/xlog.h
@@ -44,6 +44,7 @@
 
 struct iovec;
 struct xrow_header;
+struct space;
 
 #if defined(__cplusplus)
 extern "C" {
@@ -181,6 +182,18 @@ 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);
+
+/**
+ * Remove .inprogress files in vinyl subdirectories.
+ * Used as parameter to space_foreach.
+ */
+int
+collect_vinyl_inprogress(struct space *space, void *param);
+/**
  * 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..8cfeff1 100644
--- a/src/errinj.h
+++ b/src/errinj.h
@@ -111,6 +111,10 @@ 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_SNAP_MEMTX, ERRINJ_BOOL, {.bparam = false}) \
+	_(ERRINJ_VY_LOG_OPEN, ERRINJ_BOOL, {.bparam = false}) \
+	_(ERRINJ_VY_INDEX_CREATE, ERRINJ_BOOL, {.bparam = false}) \
+	_(ERRINJ_VY_RUN_COMMIT, 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..aca495e 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')
 ---
 ...
@@ -16,6 +22,8 @@ errinj.info()
     state: 0
   ERRINJ_WAL_WRITE:
     state: false
+  ERRINJ_VY_RUN_COMMIT:
+    state: false
   ERRINJ_VYRUN_DATA_READ:
     state: false
   ERRINJ_VY_SCHED_TIMEOUT:
@@ -52,10 +60,16 @@ errinj.info()
     state: false
   ERRINJ_WAL_WRITE_DISK:
     state: false
+  ERRINJ_TUPLE_FIELD:
+    state: false
   ERRINJ_VY_RUN_WRITE:
     state: false
+  ERRINJ_VY_LOG_OPEN:
+    state: false
   ERRINJ_VY_LOG_FLUSH_DELAY:
     state: false
+  ERRINJ_VY_INDEX_DUMP:
+    state: -1
   ERRINJ_SNAP_COMMIT_DELAY:
     state: false
   ERRINJ_RELAY_FINAL_SLEEP:
@@ -74,7 +88,7 @@ errinj.info()
     state: false
   ERRINJ_BUILD_SECONDARY:
     state: -1
-  ERRINJ_TUPLE_FIELD:
+  ERRINJ_VY_INDEX_CREATE:
     state: false
   ERRINJ_XLOG_GARBAGE:
     state: false
@@ -90,8 +104,8 @@ errinj.info()
     state: 0
   ERRINJ_VY_LOG_FLUSH:
     state: false
-  ERRINJ_VY_INDEX_DUMP:
-    state: -1
+  ERRINJ_SNAP_MEMTX:
+    state: false
 ...
 errinj.set("some-injection", true)
 ---
@@ -693,6 +707,157 @@ errinj.set("ERRINJ_WAL_WRITE", false)
 space:drop()
 ---
 ...
+test_run:cmd("setopt delimiter ';'")
+---
+- true
+...
+test_run:cmd("setopt delimiter ''");
+---
+- true
+...
+errinj.set("ERRINJ_SNAP_MEMTX", true)
+---
+- ok
+...
+box.snapshot()
+---
+- error: Error injection 'memtx snapshot commit'
+...
+errinj.set("ERRINJ_SNAP_MEMTX", false)
+---
+- ok
+...
+memtx_snap = string.format("%s/*.snap.inprogress", box.cfg.memtx_dir)
+---
+...
+#fio.glob(memtx_snap) > 0 -- true
+---
+- true
+...
+errinj.set("ERRINJ_VY_LOG_OPEN", true)
+---
+- ok
+...
+s = box.schema.space.create("vinyl_test", {engine='vinyl'})
+---
+...
+_ = s:create_index("primary")
+---
+...
+errinj.set("ERRINJ_VY_LOG_OPEN", false)
+---
+- ok
+...
+vinyl_vylog = string.format("%s/*.vylog.inprogress", box.cfg.vinyl_dir)
+---
+...
+#fio.glob(vinyl_vylog) > 0 -- true
+---
+- true
+...
+#fio.glob(memtx_snap) > 0 -- true
+---
+- true
+...
+vinyl_index = string.format("%s/%s/*/*.index.inprogress", box.cfg.vinyl_dir, s.id)
+---
+...
+errinj.set("ERRINJ_VY_INDEX_CREATE", 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
+---
+...
+run_dir = fio.pathjoin(box.cfg.vinyl_dir, s.id, 0)
+---
+...
+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_COMMIT", true)
+---
+- ok
+...
+while not #fio.glob(vinyl_run) == 0  do fiber.sleep(0.001) end
+---
+...
+#fio.glob(memtx_snap) > 0 -- true
+---
+- true
+...
+#fio.glob(vinyl_vylog) > 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..1bd1f05 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,63 @@ box.snapshot()
 errinj.set("ERRINJ_WAL_WRITE", false)
 space:drop()
 
+test_run:cmd("setopt delimiter ';'")
+test_run:cmd("setopt delimiter ''");
+
+errinj.set("ERRINJ_SNAP_MEMTX", true)
+box.snapshot()
+errinj.set("ERRINJ_SNAP_MEMTX", false)
+
+memtx_snap = string.format("%s/*.snap.inprogress", box.cfg.memtx_dir)
+#fio.glob(memtx_snap) > 0 -- true
+
+errinj.set("ERRINJ_VY_LOG_OPEN", true)
+s = box.schema.space.create("vinyl_test", {engine='vinyl'})
+_ = s:create_index("primary")
+errinj.set("ERRINJ_VY_LOG_OPEN", false)
+
+vinyl_vylog = string.format("%s/*.vylog.inprogress", box.cfg.vinyl_dir)
+#fio.glob(vinyl_vylog) > 0 -- true
+#fio.glob(memtx_snap) > 0 -- true
+
+vinyl_index = string.format("%s/%s/*/*.index.inprogress", box.cfg.vinyl_dir, s.id)
+errinj.set("ERRINJ_VY_INDEX_CREATE", 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
+
+run_dir = fio.pathjoin(box.cfg.vinyl_dir, s.id, 0)
+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_COMMIT", true)
+while not #fio.glob(vinyl_run) == 0  do fiber.sleep(0.001) end
+#fio.glob(memtx_snap) > 0 -- true
+#fio.glob(vinyl_vylog) > 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