[Tarantool-patches] [RFC v5 5/5] limbo: filter incoming requests

[Cyrill Gorcunov via Tarantool-patches <tarantool-patches@dev.tarantool.org>]

FIXME: This is incomplete PoC

Closes #6036

Signed-off-by: Cyrill Gorcunov <gorcunov@gmail.com>
---
 src/box/applier.cc  |  3 +++
 src/box/txn_limbo.c | 52 ++++++++++++++++++++++++++++++++++++---------
 src/box/txn_limbo.h |  9 +++++++-
 3 files changed, 53 insertions(+), 11 deletions(-)

diff --git a/src/box/applier.cc b/src/box/applier.cc
index 838aa372d..c3f3a154a 100644
--- a/src/box/applier.cc
+++ b/src/box/applier.cc
@@ -871,6 +871,9 @@ apply_synchro_row(uint32_t replica_id, struct xrow_header *row)
 		goto err;
 
 	txn_limbo_promote_lock(&txn_limbo);
+	if (txn_limbo_filter_locked(&txn_limbo, &req) != 0)
+		goto err_unlock;
+
 	struct replica_cb_data rcb_data;
 	struct synchro_entry entry;
 	/*
diff --git a/src/box/txn_limbo.c b/src/box/txn_limbo.c
index d24df3606..330ba57b2 100644
--- a/src/box/txn_limbo.c
+++ b/src/box/txn_limbo.c
@@ -731,6 +731,40 @@ txn_limbo_wait_empty(struct txn_limbo *limbo, double timeout)
 	return 0;
 }
 
+int
+txn_limbo_filter_locked(struct txn_limbo *limbo,
+			const struct synchro_request *req)
+{
+	struct txn_limbo_promote *pmt = &limbo->promote;
+	uint32_t replica_id = req->origin_id;
+	uint64_t term = req->term;
+
+	panic_on(!txn_limbo_promote_is_locked(limbo),
+		 "limbo: unlocked filtering of a request");
+
+	/*
+	 * In case of split brain has happened the promote
+	 * request may come in with already seen term.
+	 */
+	uint64_t seen_term = txn_limbo_term_locked(limbo, replica_id);
+	if (seen_term >= term) {
+		if (iproto_type_is_promote_request(req->type) &&
+		    pmt->terms_max > 1) {
+			say_info("RAFT: rejecting %s obsolete request "
+				 "from instance id %u term %llu. "
+				 "Current max term %llu.",
+				 iproto_type_name(req->type),
+				 replica_id, (long long)term,
+				 (long long)pmt->terms_max);
+			diag_set(ClientError, ER_UNSUPPORTED,
+				 "Replication", "obsolete terms");
+			return -1;
+		}
+	}
+
+	return 0;
+}
+
 void
 txn_limbo_process_locked(struct txn_limbo *limbo,
 			 const struct synchro_request *req)
@@ -742,19 +776,14 @@ txn_limbo_process_locked(struct txn_limbo *limbo,
 	panic_on(!txn_limbo_promote_is_locked(limbo),
 		 "limbo: unlocked processing of a request");
 
+	/*
+	 * Update promote tracking since bad requests must
+	 * be filtered out already.
+	 */
 	if (txn_limbo_term_locked(limbo, origin) < term) {
 		vclock_follow(&pmt->terms_map, origin, term);
 		if (term > pmt->terms_max)
 			pmt->terms_max = term;
-	} else if (iproto_type_is_promote_request(req->type) &&
-		   pmt->terms_max > 1) {
-		/* PROMOTE for outdated term. Ignore. */
-		say_info("RAFT: ignoring %s request from instance "
-			 "id %u for term %llu. Greatest term seen "
-			 "before (%llu) is bigger.",
-			 iproto_type_name(req->type), origin, (long long)term,
-			 (long long)pmt->terms_max);
-		return;
 	}
 
 	int64_t lsn = req->lsn;
@@ -800,12 +829,15 @@ txn_limbo_process_locked(struct txn_limbo *limbo,
 	return;
 }
 
-void
+int
 txn_limbo_process(struct txn_limbo *limbo, const struct synchro_request *req)
 {
 	txn_limbo_promote_lock(limbo);
+	if (txn_limbo_filter_locked(limbo, req) != 0)
+		return -1;
 	txn_limbo_process_locked(limbo, req);
 	txn_limbo_promote_unlock(limbo);
+	return 0;
 }
 
 void
diff --git a/src/box/txn_limbo.h b/src/box/txn_limbo.h
index a2595bcff..bfdfef0e0 100644
--- a/src/box/txn_limbo.h
+++ b/src/box/txn_limbo.h
@@ -358,8 +358,15 @@ txn_limbo_ack(struct txn_limbo *limbo, uint32_t replica_id, int64_t lsn);
 int
 txn_limbo_wait_complete(struct txn_limbo *limbo, struct txn_limbo_entry *entry);
 
+/**
+ * Verify if the request is valid for processing.
+ */
+int
+txn_limbo_filter_locked(struct txn_limbo *limbo,
+			const struct synchro_request *req);
+
 /** Execute a synchronous replication request. */
-void
+int
 txn_limbo_process(struct txn_limbo *limbo, const struct synchro_request *req);
 
 void
-- 
2.31.1