[PATCH] replication: recover missing local data from replica

[Konstantin Belyavskiy <k.belyavskiy@tarantool.org>]

In case of sudden power-loss, if data was not written to WAL but
already sent to remote replica, local can't recover properly and
we have different datasets. Fix it by using remote replica's data
and LSN comparison.

Based on @GeorgyKirichenko proposal.

Closes #3210
---
 This is fix for 1.6, 1.9 version was comitted earlier.
 branch: https://github.com/tarantool/tarantool/compare/gh-3210-recover-missing-local-data-master-master-16
 ticket: https://github.com/tarantool/tarantool/issues/3210
 src/box/relay.cc                               |  16 ++-
 src/box/relay.h                                |   6 ++
 test/replication/master1.lua                   |  31 ++++++
 test/replication/master2.lua                   |   1 +
 test/replication/master3.lua                   |   1 +
 test/replication/recover_missing_xlog.result   | 142 +++++++++++++++++++++++++
 test/replication/recover_missing_xlog.test.lua |  47 ++++++++
 7 files changed, 239 insertions(+), 5 deletions(-)
 create mode 100644 test/replication/master1.lua
 create mode 120000 test/replication/master2.lua
 create mode 120000 test/replication/master3.lua
 create mode 100644 test/replication/recover_missing_xlog.result
 create mode 100644 test/replication/recover_missing_xlog.test.lua

diff --git a/src/box/relay.cc b/src/box/relay.cc
index 8e53ca022..a2f7beac3 100644
--- a/src/box/relay.cc
+++ b/src/box/relay.cc
@@ -38,7 +38,6 @@
 #include "engine.h"
 #include "cluster.h"
 #include "schema.h"
-#include "vclock.h"
 #include "xrow.h"
 #include "coeio.h"
 #include "coio.h"
@@ -250,6 +249,7 @@ relay_subscribe(int fd, struct xrow_header *packet,
 	 * and identify ourselves with our own server id.
 	 */
 	struct xrow_header row;
+	vclock_copy(&relay.local_vclock_at_subscribe, master_vclock);
 	xrow_encode_vclock(&row, master_vclock);
 	/*
 	 * Identify the message with the server id of this
@@ -286,10 +286,16 @@ relay_send_row(struct recovery *r, void *param, struct xrow_header *packet)
 	 * (JOIN request). In this case, send every row.
 	 * Otherwise, we're feeding a WAL, thus responding to
 	 * SUBSCRIBE request. In that case, only send a row if
-	 * it is not from the same server (i.e. don't send
-	 * replica's own rows back).
-	 */
-	if (packet->server_id == 0 || packet->server_id != r->server_id) {
+	 * it is not from the same server (i.e. don't send replica's
+	 * own rows back) or if this row is missing on the other side
+	 * (i.e. in case of sudden power-loss, data was not written to WAL,
+	 * so remote master can't recover it). In the latter case packet's
+	 * LSN is less than or equal to local master's LSN at the moment it
+	 * received 'SUBSCRIBE' request.
+     */
+	if (packet->server_id == 0 || packet->server_id != r->server_id ||
+	    packet->lsn <= vclock_get(&relay->local_vclock_at_subscribe,
+				      packet->server_id)) {
 		relay_send(relay, packet);
 		ERROR_INJECT(ERRINJ_RELAY,
 		{
diff --git a/src/box/relay.h b/src/box/relay.h
index c13267134..b7214a41d 100644
--- a/src/box/relay.h
+++ b/src/box/relay.h
@@ -32,6 +32,7 @@
  */
 #include "evio.h"
 #include "fiber.h"
+#include "vclock.h"
 
 struct xrow_header;
 
@@ -45,6 +46,11 @@ struct relay {
 	uint64_t sync;
 	struct recovery *r;
 	ev_tstamp wal_dir_rescan_delay;
+	/**
+	 * Local vclock at the moment of subscribe, used to check
+	 * dataset on the other side and send missing data rows if any.
+	 */
+	struct vclock local_vclock_at_subscribe;
 };
 
 /**
diff --git a/test/replication/master1.lua b/test/replication/master1.lua
new file mode 100644
index 000000000..b4c054ef8
--- /dev/null
+++ b/test/replication/master1.lua
@@ -0,0 +1,31 @@
+#!/usr/bin/env tarantool
+
+-- get instance name from filename (master1.lua => master1)
+local INSTANCE_ID = string.match(arg[0], "%d")
+local USER = 'cluster'
+local PASSWORD = 'somepassword'
+local SOCKET_DIR = require('fio').cwd()
+local function instance_uri(instance_id)
+    --return 'localhost:'..(3310 + instance_id)
+    return SOCKET_DIR..'/master'..instance_id..'.sock';
+end
+
+-- start console first
+require('console').listen(os.getenv('ADMIN'))
+
+box.cfg({
+    listen = instance_uri(INSTANCE_ID);
+    replication_source = {
+        USER..':'..PASSWORD..'@'..instance_uri(1);
+        USER..':'..PASSWORD..'@'..instance_uri(2);
+        USER..':'..PASSWORD..'@'..instance_uri(3);
+    };
+})
+
+box.once("bootstrap", function()
+    local test_run = require('test_run').new()
+    box.schema.user.create(USER, { password = PASSWORD })
+    box.schema.user.grant(USER, 'replication')
+    box.schema.space.create('test', {engine = test_run:get_cfg('engine')})
+    box.space.test:create_index('primary')
+end)
diff --git a/test/replication/master2.lua b/test/replication/master2.lua
new file mode 120000
index 000000000..f6ea42dd7
--- /dev/null
+++ b/test/replication/master2.lua
@@ -0,0 +1 @@
+master1.lua
\ No newline at end of file
diff --git a/test/replication/master3.lua b/test/replication/master3.lua
new file mode 120000
index 000000000..f6ea42dd7
--- /dev/null
+++ b/test/replication/master3.lua
@@ -0,0 +1 @@
+master1.lua
\ No newline at end of file
diff --git a/test/replication/recover_missing_xlog.result b/test/replication/recover_missing_xlog.result
new file mode 100644
index 000000000..f60e5203e
--- /dev/null
+++ b/test/replication/recover_missing_xlog.result
@@ -0,0 +1,142 @@
+env = require('test_run')
+---
+...
+test_run = env.new()
+---
+...
+SERVERS = { 'master1', 'master2', 'master3' }
+---
+...
+-- Start servers
+test_run:create_cluster(SERVERS)
+---
+...
+-- Check connection status
+-- first on master 1
+test_run:cmd("switch master1")
+---
+- true
+...
+fiber = require('fiber')
+---
+...
+while box.info.replication.status ~= 'follow' do fiber.sleep(0.001) end
+---
+...
+box.info.replication.status
+---
+- follow
+...
+-- and then on master 2
+test_run:cmd("switch master2")
+---
+- true
+...
+fiber = require('fiber')
+---
+...
+while box.info.replication.status ~= 'follow' do fiber.sleep(0.001) end
+---
+...
+box.info.replication.status
+---
+- follow
+...
+-- and finally on master 3
+test_run:cmd("switch master3")
+---
+- true
+...
+fiber = require('fiber')
+---
+...
+while box.info.replication.status ~= 'follow' do fiber.sleep(0.001) end
+---
+...
+box.info.replication.status
+---
+- follow
+...
+test_run:cmd("switch master1")
+---
+- true
+...
+box.snapshot()
+---
+- ok
+...
+box.space.test:insert({1})
+---
+- [1]
+...
+box.space.test:count()
+---
+- 1
+...
+test_run:cmd("switch master3")
+---
+- true
+...
+box.space.test:count()
+---
+- 1
+...
+test_run:cmd("switch master2")
+---
+- true
+...
+box.space.test:count()
+---
+- 1
+...
+test_run:cmd("stop server master1")
+---
+- true
+...
+fio = require('fio')
+---
+...
+fio.unlink(fio.pathjoin(fio.abspath("."), string.format('master1/%020d.xlog', 0)))
+---
+- true
+...
+test_run:cmd("start server master1")
+---
+- true
+...
+test_run:cmd("switch master1")
+---
+- true
+...
+box.space.test:count()
+---
+- 1
+...
+test_run:cmd("switch default")
+---
+- true
+...
+test_run:cmd("stop server master1")
+---
+- true
+...
+test_run:cmd("stop server master2")
+---
+- true
+...
+test_run:cmd("stop server master3")
+---
+- true
+...
+test_run:cmd("cleanup server master1")
+---
+- true
+...
+test_run:cmd("cleanup server master2")
+---
+- true
+...
+test_run:cmd("cleanup server master3")
+---
+- true
+...
diff --git a/test/replication/recover_missing_xlog.test.lua b/test/replication/recover_missing_xlog.test.lua
new file mode 100644
index 000000000..ad96230ee
--- /dev/null
+++ b/test/replication/recover_missing_xlog.test.lua
@@ -0,0 +1,47 @@
+env = require('test_run')
+test_run = env.new()
+
+SERVERS = { 'master1', 'master2', 'master3' }
+-- Start servers
+test_run:create_cluster(SERVERS)
+-- Check connection status
+-- first on master 1
+test_run:cmd("switch master1")
+fiber = require('fiber')
+while box.info.replication.status ~= 'follow' do fiber.sleep(0.001) end
+box.info.replication.status
+-- and then on master 2
+test_run:cmd("switch master2")
+fiber = require('fiber')
+while box.info.replication.status ~= 'follow' do fiber.sleep(0.001) end
+box.info.replication.status
+-- and finally on master 3
+test_run:cmd("switch master3")
+fiber = require('fiber')
+while box.info.replication.status ~= 'follow' do fiber.sleep(0.001) end
+box.info.replication.status
+
+test_run:cmd("switch master1")
+box.snapshot()
+box.space.test:insert({1})
+box.space.test:count()
+
+test_run:cmd("switch master3")
+box.space.test:count()
+test_run:cmd("switch master2")
+box.space.test:count()
+test_run:cmd("stop server master1")
+fio = require('fio')
+fio.unlink(fio.pathjoin(fio.abspath("."), string.format('master1/%020d.xlog', 0)))
+test_run:cmd("start server master1")
+
+test_run:cmd("switch master1")
+box.space.test:count()
+
+test_run:cmd("switch default")
+test_run:cmd("stop server master1")
+test_run:cmd("stop server master2")
+test_run:cmd("stop server master3")
+test_run:cmd("cleanup server master1")
+test_run:cmd("cleanup server master2")
+test_run:cmd("cleanup server master3")
-- 
2.14.3 (Apple Git-98)