[PATCH 2/2] recovery: fix incorrect handling of empty-body requests.

[Serge Petrenko <sergepetrenko@tarantool.org>]

In some cases no-ops are written to xlog. They have no effect but are
needed to bump lsn. Such ops have no body, and empty body requests are
not handled in xrow_header_decode(). This leads to recovery errors in
special case: when we have a multi-statement transaction containing
no-ops written to xlog, upon recovering from such xlog, all data after
the no-op end till the start of new transaction will become no-ops body,
so, effectively, it will be ignored. Here's example `tarantoolctl cat`
output showing this (BODY contains next request data):
```
    ---
    HEADER:
      lsn: 5
      replica_id: 1
      type: NOP
      timestamp: 1536656270.5092
    BODY:
      type: 3
      timestamp: 1536656270.5092
      lsn: 6
      replica_id: 1
    ---
    HEADER:
      type: 0
    ...
```
This patch handles no-ops correctly in xrow_header_decode().

Closes #3678
---
 src/box/xrow.c                 |  6 +++-
 test/xlog/recover_nop.result   | 72 ++++++++++++++++++++++++++++++++++++++++++
 test/xlog/recover_nop.test.lua | 28 ++++++++++++++++
 3 files changed, 105 insertions(+), 1 deletion(-)
 create mode 100644 test/xlog/recover_nop.result
 create mode 100644 test/xlog/recover_nop.test.lua

diff --git a/src/box/xrow.c b/src/box/xrow.c
index 7a35d0db1..99dce5395 100644
--- a/src/box/xrow.c
+++ b/src/box/xrow.c
@@ -99,6 +99,8 @@ error:
 	if (mp_typeof(**pos) != MP_MAP)
 		goto error;
 
+	bool expect_body = true;
+
 	uint32_t size = mp_decode_map(pos);
 	for (uint32_t i = 0; i < size; i++) {
 		if (mp_typeof(**pos) != MP_UINT)
@@ -110,6 +112,8 @@ error:
 		switch (key) {
 		case IPROTO_REQUEST_TYPE:
 			header->type = mp_decode_uint(pos);
+			if (header->type == IPROTO_NOP)
+				expect_body = false;
 			break;
 		case IPROTO_SYNC:
 			header->sync = mp_decode_uint(pos);
@@ -135,7 +139,7 @@ error:
 		}
 	}
 	assert(*pos <= end);
-	if (*pos < end) {
+	if (*pos < end && expect_body) {
 		const char *body = *pos;
 		if (mp_check(pos, end)) {
 			diag_set(ClientError, ER_INVALID_MSGPACK, "packet body");
diff --git a/test/xlog/recover_nop.result b/test/xlog/recover_nop.result
new file mode 100644
index 000000000..e6aad41a7
--- /dev/null
+++ b/test/xlog/recover_nop.result
@@ -0,0 +1,72 @@
+test_run = require('test_run').new()
+---
+...
+test_run:cmd('restart server default with cleanup=1')
+-- gh-3678 check for correct recovery with nops in transaction.
+s = box.schema.space.create('test')
+---
+...
+_ = s:create_index('pk')
+---
+...
+s:replace{1,1}
+---
+- [1, 1]
+...
+f = function(old, new) return old end
+---
+...
+box.begin() s:before_replace(f) s:replace{1,2} s:before_replace(nil, f) s:replace{1,3} box.commit()
+---
+...
+s:select{}
+---
+- - [1, 3]
+...
+test_run:cmd('restart server default')
+xlog = require('xlog')
+---
+...
+fio = require('fio')
+---
+...
+box.space.test:select{}
+---
+- - [1, 3]
+...
+xlog_path = fio.pathjoin(box.cfg.wal_dir, string.format('%020d.xlog', 0))
+---
+...
+fun, param, state = xlog.pairs(xlog_path)
+---
+...
+repeat state, row = fun(param, state) until row.HEADER.type == 'NOP'
+---
+...
+row.BODY == nil
+---
+- true
+...
+state, row = fun(param, state)
+---
+...
+row.HEADER.lsn
+---
+- 6
+...
+row.HEADER.replica_id
+---
+- 1
+...
+row.HEADER.type
+---
+- REPLACE
+...
+row.BODY
+---
+- space_id: 512
+  tuple: [1, 3]
+...
+box.space.test:drop()
+---
+...
diff --git a/test/xlog/recover_nop.test.lua b/test/xlog/recover_nop.test.lua
new file mode 100644
index 000000000..de8f8ef42
--- /dev/null
+++ b/test/xlog/recover_nop.test.lua
@@ -0,0 +1,28 @@
+test_run = require('test_run').new()
+
+test_run:cmd('restart server default with cleanup=1')
+-- gh-3678 check for correct recovery with nops in transaction.
+s = box.schema.space.create('test')
+_ = s:create_index('pk')
+s:replace{1,1}
+f = function(old, new) return old end
+box.begin() s:before_replace(f) s:replace{1,2} s:before_replace(nil, f) s:replace{1,3} box.commit()
+s:select{}
+
+test_run:cmd('restart server default')
+
+xlog = require('xlog')
+fio = require('fio')
+
+box.space.test:select{}
+xlog_path = fio.pathjoin(box.cfg.wal_dir, string.format('%020d.xlog', 0))
+fun, param, state = xlog.pairs(xlog_path)
+repeat state, row = fun(param, state) until row.HEADER.type == 'NOP'
+row.BODY == nil
+state, row = fun(param, state)
+row.HEADER.lsn
+row.HEADER.replica_id
+row.HEADER.type
+row.BODY
+
+box.space.test:drop()
-- 
2.15.2 (Apple Git-101.1)