vinyl: copy key_def and range boundaries to vylog buffer

If a vylog record doesn't get flushed to disk due to an error, objects
it refers to (index->key_def, range->begin and range->end) may get
destroyed, resulting in a crash. To avoid that, we must copy those
objects to vylog buffer.

Closes #2532

src/box/vy_log.c

@@ -659,6 +659,73 @@ vy_log_record_decode(struct vy_log_record *record,
 	return -1;
 }
 
+/**
+ * Duplicate a log record. All objects refered to by the record
+ * are duplicated as well.
+ */
+static struct vy_log_record *
+vy_log_record_dup(struct mempool *pool, const struct vy_log_record *src)
+{
+	struct vy_log_record *dst = mempool_alloc(pool);
+	if (dst == NULL) {
+		diag_set(OutOfMemory, sizeof(*dst),
+			 "malloc", "struct vy_log_record");
+		goto err;
+	}
+	*dst = *src;
+	if (src->begin != NULL) {
+		const char *data = src->begin;
+		mp_next(&data);
+		size_t size = data - src->begin;
+		dst->begin = malloc(size);
+		if (dst->begin == NULL) {
+			diag_set(OutOfMemory, size, "malloc",
+				 "struct vy_log_record");
+			goto err_begin;
+		}
+		memcpy((char *)dst->begin, src->begin, size);
+	}
+	if (src->end != NULL) {
+		const char *data = src->end;
+		mp_next(&data);
+		size_t size = data - src->end;
+		dst->end = malloc(size);
+		if (dst->end == NULL) {
+			diag_set(OutOfMemory, size, "malloc",
+				 "struct vy_log_record");
+			goto err_end;
+		}
+		memcpy((char *)dst->end, src->end, size);
+	}
+	if (src->key_def != NULL) {
+		dst->key_def = key_def_dup(src->key_def);
+		if (dst->key_def == NULL)
+			goto err_key_def;
+	}
+	return dst;
+
+err_key_def:
+	free((char *)dst->end);
+err_end:
+	free((char *)dst->begin);
+err_begin:
+	mempool_free(pool, dst);
+err:
+	return NULL;
+}
+
+/**
+ * Free a log record and all objects it refers to.
+ */
+static void
+vy_log_record_free(struct mempool *pool, struct vy_log_record *record)
+{
+	free((char *)record->begin);
+	free((char *)record->end);
+	free((struct key_def *)record->key_def);
+	mempool_free(pool, record);
+}
+
 void
 vy_log_init(const char *dir)
 {
@@ -725,7 +792,7 @@ vy_log_flush(void)
 	/* Success. Free flushed records. */
 	struct vy_log_record *next;
 	stailq_foreach_entry_safe(record, next, &vy_log.tx, in_tx)
-		mempool_free(&vy_log.record_pool, record);
+		vy_log_record_free(&vy_log.record_pool, record);
 	stailq_create(&vy_log.tx);
 	vy_log.tx_size = 0;
 	return 0;
@@ -1132,7 +1199,7 @@ vy_log_tx_do_commit(bool no_discard)
 	stailq_foreach_entry_safe(record, next, &rollback, in_tx) {
 		assert(vy_log.tx_size > 0);
 		vy_log.tx_size--;
-		mempool_free(&vy_log.record_pool, record);
+		vy_log_record_free(&vy_log.record_pool, record);
 	}
 	goto out;
 }
@@ -1156,16 +1223,14 @@ vy_log_write(const struct vy_log_record *record)
 
 	say_debug("%s: %s", __func__, vy_log_record_str(record));
 
-	struct vy_log_record *tx_record = mempool_alloc(&vy_log.record_pool);
+	struct vy_log_record *tx_record;
+	tx_record = vy_log_record_dup(&vy_log.record_pool, record);
 	if (tx_record == NULL) {
-		diag_set(OutOfMemory, sizeof(*tx_record),
-			 "malloc", "struct vy_log_record");
 		diag_move(diag_get(), &vy_log.tx_diag);
 		vy_log.tx_failed = true;
 		return;
 	}
 
-	*tx_record = *record;
 	stailq_add_tail_entry(&vy_log.tx, tx_record, in_tx);
 	vy_log.tx_size++;
 	if (vy_log.tx_begin == NULL)

test/vinyl/errinj_vylog.result

@@ -219,6 +219,22 @@ _ = s3:create_index('pk')
 _ = s3:insert{666, 'fff'}
 ---
 ...
+-- gh-2532: replaying create/drop from xlog crashes tarantool
+test_run:cmd("setopt delimiter ';'")
+---
+- true
+...
+for i = 1, 10 do
+    s = box.schema.space.create('test', {engine = 'vinyl'})
+    s:create_index('primary')
+    s:create_index('secondary', {unique = false, parts = {2, 'string'}})
+    s:insert{i, 'test' .. i}
+    s:truncate()
+    s:drop()
+end
+test_run:cmd("setopt delimiter ''");
+---
+...
 test_run:cmd('restart server default')
 s1 = box.space.test1
 ---

test/vinyl/errinj_vylog.test.lua

@@ -107,6 +107,18 @@ s3 = box.schema.space.create('test3', {engine = 'vinyl'})
 _ = s3:create_index('pk')
 _ = s3:insert{666, 'fff'}
 
+-- gh-2532: replaying create/drop from xlog crashes tarantool
+test_run:cmd("setopt delimiter ';'")
+for i = 1, 10 do
+    s = box.schema.space.create('test', {engine = 'vinyl'})
+    s:create_index('primary')
+    s:create_index('secondary', {unique = false, parts = {2, 'string'}})
+    s:insert{i, 'test' .. i}
+    s:truncate()
+    s:drop()
+end
+test_run:cmd("setopt delimiter ''");
+
 test_run:cmd('restart server default')
 
 s1 = box.space.test1