[PATCH v2 1/6] xlog: store prev vclock in xlog header

[Vladimir Davydov <vdavydov.dev@gmail.com>]

This patch adds a new key to xlog header, PrevVclock, which contains the
vclock of the previous xlog file in the xlog directory. It is set by
xdir_create_xlog() to the last vclock in xdir::index. The new key is
only present in XLOG files (it doesn't make sense for SNAP or VYLOG
anyway). It will be used to make the check for xlog gaps more thorough.
---
 src/box/xlog.c            | 70 ++++++++++++++++++++++++++++++++++++-----------
 src/box/xlog.h            |  8 ++++++
 test/xlog/header.result   | 41 +++++++++++++++++++++++++++
 test/xlog/header.test.lua | 13 +++++++++
 4 files changed, 116 insertions(+), 16 deletions(-)

diff --git a/src/box/xlog.c b/src/box/xlog.c
index 142262b4..db35f474 100644
--- a/src/box/xlog.c
+++ b/src/box/xlog.c
@@ -94,6 +94,7 @@ enum {
 #define INSTANCE_UUID_KEY_V12 "Server"
 #define VCLOCK_KEY "VClock"
 #define VERSION_KEY "Version"
+#define PREV_VCLOCK_KEY "PrevVClock"
 
 static const char v13[] = "0.13";
 static const char v12[] = "0.12";
@@ -117,19 +118,50 @@ xlog_meta_format(const struct xlog_meta *meta, char *buf, int size)
 	if (vstr == NULL)
 		return -1;
 	char *instance_uuid = tt_uuid_str(&meta->instance_uuid);
-	int total = snprintf(buf, size,
+	int total = 0;
+	SNPRINT(total, snprintf, buf, size,
 		"%s\n"
 		"%s\n"
 		VERSION_KEY ": %s\n"
 		INSTANCE_UUID_KEY ": %s\n"
-		VCLOCK_KEY ": %s\n\n",
+		VCLOCK_KEY ": %s\n",
 		meta->filetype, v13, PACKAGE_VERSION, instance_uuid, vstr);
-	assert(total > 0);
 	free(vstr);
+	if (meta->has_prev_vclock) {
+		vstr = vclock_to_string(&meta->prev_vclock);
+		SNPRINT(total, snprintf, buf, size,
+			PREV_VCLOCK_KEY ": %s\n", vstr);
+		free(vstr);
+	}
+	SNPRINT(total, snprintf, buf, size, "\n");
+	assert(total > 0);
 	return total;
 }
 
 /**
+ * Parse vclock from xlog meta.
+ */
+static int
+parse_vclock(const char *val, const char *val_end, struct vclock *vclock)
+{
+	if (val_end - val > VCLOCK_STR_LEN_MAX) {
+		diag_set(XlogError, "can't parse vclock");
+		return -1;
+	}
+	char str[VCLOCK_STR_LEN_MAX + 1];
+	memcpy(str, val, val_end - val);
+	str[val_end - val] = '\0';
+	size_t off = vclock_from_string(vclock, str);
+	ERROR_INJECT(ERRINJ_XLOG_META, { off = 1; });
+	if (off != 0) {
+		diag_set(XlogError, "invalid vclock at "
+			 "offset %zd", off);
+		return -1;
+	}
+	return 0;
+}
+
+/**
  * Parse xlog meta from buffer, update buffer read
  * position in case of success
  *
@@ -224,21 +256,15 @@ xlog_meta_parse(struct xlog_meta *meta, const char **data,
 			/*
 			 * VClock: <vclock>
 			 */
-			if (val_end - val > VCLOCK_STR_LEN_MAX) {
-				diag_set(XlogError, "can't parse vclock");
+			if (parse_vclock(val, val_end, &meta->vclock) != 0)
 				return -1;
-			}
-			char vclock[VCLOCK_STR_LEN_MAX + 1];
-			memcpy(vclock, val, val_end - val);
-			vclock[val_end - val] = '\0';
-			size_t off = vclock_from_string(&meta->vclock, vclock);
-			ERROR_INJECT(ERRINJ_XLOG_META, {
-				off = 1;});
-			if (off != 0) {
-				diag_set(XlogError, "invalid vclock at "
-					  "offset %zd", off);
+		} else if (memcmp(key, PREV_VCLOCK_KEY, key_end - key) == 0) {
+			/*
+			 * PrevVClock: <vclock>
+			 */
+			if (parse_vclock(val, val_end, &meta->prev_vclock) != 0)
 				return -1;
-			}
+			meta->has_prev_vclock = true;
 		} else if (memcmp(key, VERSION_KEY, key_end - key) == 0) {
 			/* Ignore Version: for now */
 		} else {
@@ -868,6 +894,18 @@ xdir_create_xlog(struct xdir *dir, struct xlog *xlog,
 	meta.instance_uuid = *dir->instance_uuid;
 	vclock_copy(&meta.vclock, vclock);
 
+	/*
+	 * For WAL dir: store vclock of the previous xlog file
+	 * to check for gaps on recovery.
+	 */
+	if (dir->type == XLOG && !vclockset_empty(&dir->index)) {
+		vclock_copy(&meta.prev_vclock, vclockset_last(&dir->index));
+		meta.has_prev_vclock = true;
+	} else {
+		vclock_create(&meta.prev_vclock);
+		meta.has_prev_vclock = false;
+	}
+
 	if (xlog_create(xlog, filename, dir->open_wflags, &meta) != 0)
 		return -1;
 
diff --git a/src/box/xlog.h b/src/box/xlog.h
index 4b9a5765..bff3275f 100644
--- a/src/box/xlog.h
+++ b/src/box/xlog.h
@@ -229,6 +229,14 @@ struct xlog_meta {
 	 * is vector clock *at the time the snapshot is taken*.
 	 */
 	struct vclock vclock;
+	/**
+	 * Text file header: vector clock of the previous
+	 * file at the directory. Used for checking the
+	 * directory for missing WALs.
+	 */
+	struct vclock prev_vclock;
+	/** Set if @prev_vclock is present. */
+	bool has_prev_vclock;
 };
 
 /* }}} */
diff --git a/test/xlog/header.result b/test/xlog/header.result
index 4b8431e8..6520e469 100644
--- a/test/xlog/header.result
+++ b/test/xlog/header.result
@@ -103,6 +103,47 @@ dump_header(fio.pathjoin(box.cfg.wal_dir, xlog_name))
   - 'Version: <version>'
   - 'Instance: <instance_uuid>'
   - 'VClock: {1: 2}'
+  - 'PrevVClock: {}'
+...
+box.space._schema:delete({"layout_test"})
+---
+- ['layout_test']
+...
+box.snapshot()
+---
+- ok
+...
+checkpoint_lsn = box.info.lsn
+---
+...
+-- SNAP files
+snap_name = string.format("%020d.snap", checkpoint_lsn)
+---
+...
+dump_header(fio.pathjoin(box.cfg.memtx_dir, snap_name))
+---
+- - SNAP
+  - '0.13'
+  - 'Version: <version>'
+  - 'Instance: <instance_uuid>'
+  - 'VClock: {1: 4}'
+...
+-- XLOG files
+box.space._schema:insert({"layout_test"})
+---
+- ['layout_test']
+...
+xlog_name = string.format("%020d.xlog", checkpoint_lsn)
+---
+...
+dump_header(fio.pathjoin(box.cfg.wal_dir, xlog_name))
+---
+- - XLOG
+  - '0.13'
+  - 'Version: <version>'
+  - 'Instance: <instance_uuid>'
+  - 'VClock: {1: 4}'
+  - 'PrevVClock: {1: 2}'
 ...
 box.space._schema:delete({"layout_test"})
 ---
diff --git a/test/xlog/header.test.lua b/test/xlog/header.test.lua
index 1e7ec794..686d2bed 100644
--- a/test/xlog/header.test.lua
+++ b/test/xlog/header.test.lua
@@ -44,4 +44,17 @@ xlog_name = string.format("%020d.xlog", checkpoint_lsn)
 dump_header(fio.pathjoin(box.cfg.wal_dir, xlog_name))
 box.space._schema:delete({"layout_test"})
 
+box.snapshot()
+checkpoint_lsn = box.info.lsn
+
+-- SNAP files
+snap_name = string.format("%020d.snap", checkpoint_lsn)
+dump_header(fio.pathjoin(box.cfg.memtx_dir, snap_name))
+
+-- XLOG files
+box.space._schema:insert({"layout_test"})
+xlog_name = string.format("%020d.xlog", checkpoint_lsn)
+dump_header(fio.pathjoin(box.cfg.wal_dir, xlog_name))
+box.space._schema:delete({"layout_test"})
+
 test_run:cmd("clear filter")
-- 
2.11.0