[PATCH] replication: automatic skip duplicating rows in replication

[Vladimir Davydov]

On Thu, Apr 12, 2018 at 02:01:25PM +0300, Konstantin Belyavskiy wrote:
> ticket: https://github.com/tarantool/tarantool/issues/3270
> branch: https://github.com/tarantool/tarantool/compare/gh-3270-add-skip-conflict-row-option
> 
> In case of attempting to insert a duplicate key, an error ER_TUPLE_FOUND
> occured, which led to disconnect. 
> Introduce new oftion: 'silent_skip_conflict_rows', if set, then error of
> this type will be ignored.
> 
> Closes #3270
> ---
>  src/box/applier.cc                          |  15 +++-
>  src/box/lua/load_cfg.lua                    |   3 +
>  test/app-tap/init_script.result             |  39 +++++----
>  test/box/admin.result                       |   2 +
>  test/box/cfg.result                         |   4 +
>  test/replication/replica_skip_row.lua       |  10 +++
>  test/replication/skip_conflict_row.result   | 129 ++++++++++++++++++++++++++++
>  test/replication/skip_conflict_row.test.lua |  46 ++++++++++
>  8 files changed, 228 insertions(+), 20 deletions(-)
>  create mode 100644 test/replication/replica_skip_row.lua
>  create mode 100644 test/replication/skip_conflict_row.result
>  create mode 100644 test/replication/skip_conflict_row.test.lua
> 
> diff --git a/src/box/applier.cc b/src/box/applier.cc
> index 9aa951c34..5ebe67e27 100644
> --- a/src/box/applier.cc
> +++ b/src/box/applier.cc
> @@ -47,6 +47,7 @@
>  #include "xrow_io.h"
>  #include "error.h"
>  #include "session.h"
> +#include "cfg.h"
>  
>  STRS(applier_state, applier_STATE);
>  
> @@ -505,7 +506,19 @@ applier_subscribe(struct applier *applier)
>  			 */
>  			vclock_follow(&replicaset.vclock, row.replica_id,
>  				      row.lsn);
> -			xstream_write_xc(applier->subscribe_stream, &row);
> +			if (xstream_write(applier->subscribe_stream, &row) != 0) {
> +				struct error *e = diag_last_error(diag_get());
> +				/**
> +				 * Silently skip ER_TUPLE_FOUND error if such
> +				 * option is set in config.
> +				 */
> +				if (e->type == &type_ClientError &&
> +				    box_error_code(e) == ER_TUPLE_FOUND &&

> +				    cfg_geti("silent_skip_conflict_rows"))

cfg_geti() is a heavy operation. I think you should cache the value of
this configuration option in C.

> +					diag_clear(diag_get());
> +				else
> +					diag_raise();
> +			}

What about UPDATE over a non-existent key or vinyl transaction conflict?
Those can also happen due to replication conflicts. May be, we should
ignore all errors of type ClientError?

>  		}
>  		if (applier->state == APPLIER_SYNC ||
>  		    applier->state == APPLIER_FOLLOW)
> diff --git a/src/box/lua/load_cfg.lua b/src/box/lua/load_cfg.lua
> index 89fd7745e..372f00f25 100644
> --- a/src/box/lua/load_cfg.lua
> +++ b/src/box/lua/load_cfg.lua
> @@ -62,6 +62,7 @@ local default_cfg = {
>      feedback_enabled      = true,
>      feedback_host         = "https://feedback.tarantool.io",
>      feedback_interval     = 3600,
> +    silent_skip_conflict_rows = false,

All replication-related options should have replication_ prefix.
What about replication_ignore_conflicts?

> diff --git a/test/replication/skip_conflict_row.test.lua b/test/replication/skip_conflict_row.test.lua
> new file mode 100644
> index 000000000..5cfbff339
> --- /dev/null
> +++ b/test/replication/skip_conflict_row.test.lua
> @@ -0,0 +1,46 @@
> +env = require('test_run')
> +test_run = env.new()
> +engine = test_run:get_cfg('engine')
> +
> +box.schema.user.grant('guest', 'read,write,execute', 'universe')
> +box.schema.user.grant('guest', 'replication')
> +
> +space = box.schema.space.create('test', {engine = engine});
> +index = box.space.test:create_index('primary')
> +
> +test_run:cmd("create server replica with rpl_master=default, script='replication/replica_skip_row.lua'")

AFAIU you don't need to add a new script - you can reuse replica.lua and
set the new configuration option after bootstrap.

> +test_run:cmd("start server replica")
> +
> +repl = box.cfg.replication
> +box.cfg{replication = ""}
> +box.info.status

I don't understand why you turn off replication here.

> +
> +test_run:cmd("switch replica")
> +repl = box.cfg.replication
> +box.cfg{replication = ""}

... and here

> +box.space.test:insert{2}
> +box.space.test:insert{1}
> +
> +test_run:cmd("switch default")
> +space:insert{1}
> +space:select{}
> +box.cfg{replication = repl}
> +box.info.status
> +
> +test_run:cmd("switch replica")
> +box.cfg{replication = repl}
> +require('fiber').sleep(0.01)
> +box.info.replication[1].upstream.message
> +box.info.replication[1].upstream.status
> +box.space.test:select{}
> +
> +test_run:cmd("switch default")
> +space:select{}
> +box.info.status
> +
> +-- cleanup
> +test_run:cmd("stop server replica")
> +test_run:cmd("cleanup server replica")
> +box.space.test:drop()
> +box.schema.user.revoke('guest', 'replication')
> +box.schema.user.revoke('guest', 'read,write,execute', 'universe')