[Tarantool-patches] [PATCH v2 12/19] replication: support ROLLBACK and CONFIRM during recovery

[Vladislav Shpilevoy]

From: Serge Petrenko <sergepetrenko at tarantool.org>

Follow-up #4847
Follow-up #4848
---
 src/box/box.cc      | 20 ++++++++++++++++++--
 src/box/txn_limbo.c |  6 ++----
 2 files changed, 20 insertions(+), 6 deletions(-)

diff --git a/src/box/box.cc b/src/box/box.cc
index 32c69df76..ad76f4f00 100644
--- a/src/box/box.cc
+++ b/src/box/box.cc
@@ -373,9 +373,25 @@ static void
 apply_wal_row(struct xstream *stream, struct xrow_header *row)
 {
 	struct request request;
-	// TODO: process confirmation during recovery.
-	if (iproto_type_is_synchro_request(row->type))
+	if (iproto_type_is_synchro_request(row->type)) {
+		uint32_t replica_id;
+		int64_t lsn;
+		switch(row->type) {
+		case IPROTO_CONFIRM:
+			if (xrow_decode_confirm(row, &replica_id, &lsn) < 0)
+				diag_raise();
+			assert(txn_limbo.instance_id == replica_id);
+			txn_limbo_read_confirm(&txn_limbo, lsn);
+			break;
+		case IPROTO_ROLLBACK:
+			if (xrow_decode_rollback(row, &replica_id, &lsn) < 0)
+				diag_raise();
+			assert(txn_limbo.instance_id == replica_id);
+			txn_limbo_read_rollback(&txn_limbo, lsn);
+			break;
+		}
 		return;
+	}
 	xrow_decode_dml_xc(row, &request, dml_request_key_map(row->type));
 	if (request.type != IPROTO_NOP) {
 		struct space *space = space_cache_find_xc(request.space_id);
diff --git a/src/box/txn_limbo.c b/src/box/txn_limbo.c
index 680e81d3d..d3751a28b 100644
--- a/src/box/txn_limbo.c
+++ b/src/box/txn_limbo.c
@@ -230,8 +230,7 @@ txn_limbo_write_confirm(struct txn_limbo *limbo,
 void
 txn_limbo_read_confirm(struct txn_limbo *limbo, int64_t lsn)
 {
-	assert(limbo->instance_id != REPLICA_ID_NIL &&
-	       limbo->instance_id != instance_id);
+	assert(limbo->instance_id != REPLICA_ID_NIL);
 	struct txn_limbo_entry *e, *tmp;
 	rlist_foreach_entry_safe(e, &limbo->queue, in_queue, tmp) {
 		if (e->lsn > lsn)
@@ -265,8 +264,7 @@ txn_limbo_write_rollback(struct txn_limbo *limbo,
 void
 txn_limbo_read_rollback(struct txn_limbo *limbo, int64_t lsn)
 {
-	assert(limbo->instance_id != REPLICA_ID_NIL &&
-	       limbo->instance_id != instance_id);
+	assert(limbo->instance_id != REPLICA_ID_NIL);
 	struct txn_limbo_entry *e, *tmp;
 	rlist_foreach_entry_safe_reverse(e, &limbo->queue, in_queue, tmp) {
 		if (e->lsn <= lsn)
-- 
2.21.1 (Apple Git-122.3)