[tarantool-patches] [PATCH] Raise an error if remote transaction produces non-local changes

[Georgy Kirichenko <georgy@tarantool.org>]

Disallow changes for non-local spaces during replication stream
applying. As we do not support distributed transaction yet we could not
provide a transactional replication for such side effects if there are
not NOPed.

Needed for: #2798
Follow up for: 27283debc327a1ef87e025badeed97d9ac264ac6

Branch: https://github.com/tarantool/tarantool/tree/g.kirichenko/gh-2798-replication-disable-non-local-changes
---
 src/box/txn.c                        |  7 +++++
 test/replication/on_replace.result   | 44 ++++++++++++++++++++++++----
 test/replication/on_replace.test.lua | 20 +++++++++++--
 3 files changed, 63 insertions(+), 8 deletions(-)

diff --git a/src/box/txn.c b/src/box/txn.c
index deb4fac47..40ba3aaf4 100644
--- a/src/box/txn.c
+++ b/src/box/txn.c
@@ -244,6 +244,13 @@ txn_commit_stmt(struct txn *txn, struct request *request)
 		else
 			++txn->n_remote_rows;
 	}
+	if (txn->n_remote_rows > 0 && stmt->row->replica_id == 0 &&
+	    stmt->space->def->opts.group_id != GROUP_LOCAL) {
+		diag_set(ClientError, ER_UNSUPPORTED,
+			 "transaction", "mixing local and remote changes");
+		goto fail;
+	}
+
 	/*
 	 * If there are triggers, and they are not disabled, and
 	 * the statement found any rows, run triggers.
diff --git a/test/replication/on_replace.result b/test/replication/on_replace.result
index 2e95b90ea..3bffefc3d 100644
--- a/test/replication/on_replace.result
+++ b/test/replication/on_replace.result
@@ -104,7 +104,7 @@ box.space.test:drop()
 box.schema.user.revoke('guest', 'replication')
 ---
 ...
--- gh-2682 on_replace on slave server with data change
+-- gh-2798 on_replace on slave server with non-local data change should fail
 SERVERS = { 'on_replace1', 'on_replace2' }
 ---
 ...
@@ -143,7 +143,7 @@ fiber = require'fiber'
 while box.space.s2 == nil do fiber.sleep(0.00001) end
 ---
 ...
-_ = box.space.s1:on_replace(function (old, new) box.space.s2:replace(new) end)
+tg = box.space.s1:on_replace(function (old, new) box.space.s2:replace(new) end)
 ---
 ...
 test_run:cmd('switch on_replace1')
@@ -154,20 +154,27 @@ box.space.s1:replace({1, 2, 3, 4})
 ---
 - [1, 2, 3, 4]
 ...
-while #(box.space.s2:select()) == 0 do fiber.sleep(0.00001) end
+while (box.info.replication[3 - box.info.id].downstream.status ~= 'stopped') do fiber.sleep(0.00001) end
 ---
 ...
 test_run:cmd('switch on_replace2')
 ---
 - true
 ...
+while (box.info.replication[3 - box.info.id].upstream.status ~= 'stopped') do fiber.sleep(0.00001) end
+---
+...
+box.info.replication[3 - box.info.id].upstream.message
+---
+- transaction does not support mixing local and remote changes
+...
 box.space.s1:select()
 ---
-- - [1, 2, 3, 4]
+- []
 ...
 box.space.s2:select()
 ---
-- - [1, 2, 3, 4]
+- []
 ...
 test_run:cmd('switch on_replace1')
 ---
@@ -179,6 +186,33 @@ box.space.s1:select()
 ...
 box.space.s2:select()
 ---
+- []
+...
+-- gh-2798 on_replace on slave server with local data change is allowed
+test_run:cmd('switch on_replace2')
+---
+- true
+...
+s3 = box.schema.space.create('s3', {is_local = true})
+---
+...
+_ = s3:create_index('pk')
+---
+...
+tg = box.space.s1:on_replace(function (old, new) box.space.s3:replace(new) end, tg)
+---
+...
+replication = box.cfg.replication
+---
+...
+box.cfg{replication = {}}
+---
+...
+box.cfg{replication = replication}
+---
+...
+s3:select()
+---
 - - [1, 2, 3, 4]
 ...
 _ = test_run:cmd('switch default')
diff --git a/test/replication/on_replace.test.lua b/test/replication/on_replace.test.lua
index e34832103..779dbf768 100644
--- a/test/replication/on_replace.test.lua
+++ b/test/replication/on_replace.test.lua
@@ -44,7 +44,7 @@ box.space.test:drop()
 box.schema.user.revoke('guest', 'replication')
 
 
--- gh-2682 on_replace on slave server with data change
+-- gh-2798 on_replace on slave server with non-local data change should fail
 
 SERVERS = { 'on_replace1', 'on_replace2' }
 test_run:create_cluster(SERVERS, "replication", {args="0.2"})
@@ -60,13 +60,15 @@ _ = s2:create_index('pk')
 test_run:cmd('switch on_replace2')
 fiber = require'fiber'
 while box.space.s2 == nil do fiber.sleep(0.00001) end
-_ = box.space.s1:on_replace(function (old, new) box.space.s2:replace(new) end)
+tg = box.space.s1:on_replace(function (old, new) box.space.s2:replace(new) end)
 
 test_run:cmd('switch on_replace1')
 box.space.s1:replace({1, 2, 3, 4})
-while #(box.space.s2:select()) == 0 do fiber.sleep(0.00001) end
+while (box.info.replication[3 - box.info.id].downstream.status ~= 'stopped') do fiber.sleep(0.00001) end
 
 test_run:cmd('switch on_replace2')
+while (box.info.replication[3 - box.info.id].upstream.status ~= 'stopped') do fiber.sleep(0.00001) end
+box.info.replication[3 - box.info.id].upstream.message
 box.space.s1:select()
 box.space.s2:select()
 
@@ -74,6 +76,18 @@ test_run:cmd('switch on_replace1')
 box.space.s1:select()
 box.space.s2:select()
 
+-- gh-2798 on_replace on slave server with local data change is allowed
+test_run:cmd('switch on_replace2')
+s3 = box.schema.space.create('s3', {is_local = true})
+_ = s3:create_index('pk')
+tg = box.space.s1:on_replace(function (old, new) box.space.s3:replace(new) end, tg)
+
+replication = box.cfg.replication
+box.cfg{replication = {}}
+box.cfg{replication = replication}
+
+s3:select()
+
 _ = test_run:cmd('switch default')
 test_run:drop_cluster(SERVERS)
 test_run:cleanup_cluster()
-- 
2.21.0