Re: [Tarantool-patches] [PATCH 6/9] raft: keep track of greatest known term and filter replication sources based on that

[Serge Petrenko via Tarantool-patches <tarantool-patches@dev.tarantool.org>]

11.04.2021 20:56, Serge Petrenko пишет:
> Start writing the actual leader term together with the PROMOTE request
> and process terms in PROMOTE requests on receiver side.
>
> Make applier only apply synchronous transactions from the instance which
> has the greatest term as received in PROMOTE requests.
>
> Closes #5445
> ---

Force-pushed a test and a couple of fixes. Please find an incremental 
diff below and
the full patch in v2.

==========================
diff --git a/src/box/box.cc b/src/box/box.cc
index 9b6323b3f..aae57ec29 100644
--- a/src/box/box.cc
+++ b/src/box/box.cc
@@ -1507,7 +1507,12 @@ box_clear_synchro_queue(bool try_wait)
          return -1;
      }

-    if (!is_box_configured)
+    /*
+     * Do nothing when box isn't configured and when PROMOTE was already
+     * written for this term.
+     */
+    if (!is_box_configured ||
+        raft_source_term(box_raft(), instance_id) == box_raft()->term)
          return 0;
      uint32_t former_leader_id = txn_limbo.owner_id;
      int64_t wait_lsn = txn_limbo.confirmed_lsn;
@@ -1569,12 +1574,12 @@ promote:
                  .replica_id = 0, /* unused */
                  .origin_id = instance_id,
                  .lsn = wait_lsn,
-                .term = 0, /* unused */
+                .term = box_raft()->term,
              };
              txn_limbo_read_promote(&txn_limbo, &req);
              assert(txn_limbo_is_empty(&txn_limbo));
              raft_source_update_term(box_raft(), req.origin_id,
-                        req.lsn);
+                        req.term);
          }
      }
      in_clear_synchro_queue = false;
diff --git a/src/lib/raft/raft.h b/src/lib/raft/raft.h
index cba45a67d..40c8630e9 100644
--- a/src/lib/raft/raft.h
+++ b/src/lib/raft/raft.h
@@ -256,20 +256,29 @@ raft_is_source_allowed(const struct raft *raft, 
uint32_t source_id)
      return !raft->is_enabled || raft->leader == source_id;
  }

-static inline bool
-raft_source_has_outdated_term(const struct raft *raft, uint32_t source_id)
+/**
+ * Return the latest term as seen in PROMOTE requests from instance with id
+ * @a source_id.
+ */
+static inline uint64_t
+raft_source_term(const struct raft *raft, uint32_t source_id)
  {
-    uint64_t source_term = vclock_get(&raft->term_map, source_id);
-    return raft->is_enabled && source_term < raft->greatest_known_term;
+    assert(source_id != 0 && source_id < VCLOCK_MAX);
+    return vclock_get(&raft->term_map, source_id);
  }

-/** Check if Raft is enabled. */
+/**
+ * Check whether replica with id @a source_id is too old to apply 
synchronous
+ * data from it. The check remains valid  even when elections are disabled.
+ */
  static inline bool
-raft_is_enabled(const struct raft *raft)
+raft_source_has_outdated_term(const struct raft *raft, uint32_t source_id)
  {
-    return raft->is_enabled;
+    uint64_t source_term = vclock_get(&raft->term_map, source_id);
+    return source_term < raft->greatest_known_term;
  }

+/** Remember the last term seen for replica  with id @a source_id. */
  static inline void
  raft_source_update_term(struct raft *raft, uint32_t source_id, 
uint64_t term)
  {
@@ -280,6 +289,13 @@ raft_source_update_term(struct raft *raft, uint32_t 
source_id, uint64_t term)
          raft->greatest_known_term = term;
  }

+/** Check if Raft is enabled. */
+static inline bool
+raft_is_enabled(const struct raft *raft)
+{
+    return raft->is_enabled;
+}
+
  /** Process a raft entry stored in WAL/snapshot. */
  void
  raft_process_recovery(struct raft *raft, const struct raft_msg *req);
diff --git a/test/replication/gh-5445-leader-incosistency.result 
b/test/replication/gh-5445-leader-incosistency.result
new file mode 100644
index 000000000..b1f8a4ed1
--- /dev/null
+++ b/test/replication/gh-5445-leader-incosistency.result
@@ -0,0 +1,238 @@
+-- test-run result file version 2
+test_run = require("test_run").new()
+ | ---
+ | ...
+
+is_leader_cmd = "return box.info.election.state == 'leader'"
+ | ---
+ | ...
+
+-- Auxiliary.
+test_run:cmd('setopt delimiter ";"')
+ | ---
+ | - true
+ | ...
+function get_leader(nrs)
+    local leader_nr = 0
+    test_run:wait_cond(function()
+        for nr, do_check in pairs(nrs) do
+            if do_check then
+                local is_leader = test_run:eval('election_replica'..nr,
+                                                is_leader_cmd)[1]
+                if is_leader then
+                    leader_nr = nr
+                    return true
+                end
+            end
+        end
+        return false
+    end)
+    assert(leader_nr ~= 0)
+    return leader_nr
+end;
+ | ---
+ | ...
+
+function name(id)
+    return 'election_replica'..id
+end;
+ | ---
+ | ...
+test_run:cmd('setopt delimiter ""');
+ | ---
+ | - true
+ | ...
+
+--
+-- gh-5445: make sure rolled back rows do not reappear once old leader 
returns
+-- to cluster.
+--
+SERVERS = {'election_replica1', 'election_replica2' ,'election_replica3'}
+ | ---
+ | ...
+test_run:create_cluster(SERVERS, "replication", {args='2 0.4'})
+ | ---
+ | ...
+test_run:wait_fullmesh(SERVERS)
+ | ---
+ | ...
+
+-- Any of the three instances may bootstrap the cluster and become leader.
+is_possible_leader = {true, true, true}
+ | ---
+ | ...
+leader_nr = get_leader(is_possible_leader)
+ | ---
+ | ...
+leader = name(leader_nr)
+ | ---
+ | ...
+next_leader_nr = ((leader_nr - 1) % 3 + 1) % 3 + 1 -- {1, 2, 3} -> {2, 
3, 1}
+ | ---
+ | ...
+next_leader = name(next_leader_nr)
+ | ---
+ | ...
+other_nr = ((leader_nr - 1) % 3 + 2) % 3 + 1 -- {1, 2, 3} -> {3, 1, 2}
+ | ---
+ | ...
+other = name(other_nr)
+ | ---
+ | ...
+
+test_run:switch(leader)
+ | ---
+ | - true
+ | ...
+box.ctl.wait_rw()
+ | ---
+ | ...
+_ = box.schema.space.create('test', {is_sync=true})
+ | ---
+ | ...
+_ = box.space.test:create_index('pk')
+ | ---
+ | ...
+box.space.test:insert{1}
+ | ---
+ | - [1]
+ | ...
+
+-- Simulate a situation when the instance which will become the next leader
+-- doesn't know of unconfirmed rows. It should roll them back anyways 
and do not
+-- accept them once they actually appear from the old leader.
+-- So, stop the instance which'll be the next leader.
+test_run:switch('default')
+ | ---
+ | - true
+ | ...
+test_run:cmd('stop server '..next_leader)
+ | ---
+ | - true
+ | ...
+test_run:switch(leader)
+ | ---
+ | - true
+ | ...
+-- Insert some unconfirmed data.
+box.cfg{replication_synchro_quorum=3, replication_synchro_timeout=1000}
+ | ---
+ | ...
+fib = require('fiber').create(box.space.test.insert, box.space.test, {2})
+ | ---
+ | ...
+fib:status()
+ | ---
+ | - suspended
+ | ...
+
+-- 'other', 'leader', 'next_leader' are defined on 'default' node, 
hence the
+-- double switches.
+test_run:switch('default')
+ | ---
+ | - true
+ | ...
+test_run:switch(other)
+ | ---
+ | - true
+ | ...
+-- Wait until the rows are replicated to the other instance.
+test_run:wait_cond(function() return box.space.test:get{2} ~= nil end)
+ | ---
+ | - true
+ | ...
+box.cfg{election_mode='voter'}
+ | ---
+ | ...
+-- Old leader is gone.
+test_run:switch('default')
+ | ---
+ | - true
+ | ...
+test_run:cmd('stop server '..leader)
+ | ---
+ | - true
+ | ...
+is_possible_leader[leader_nr] = false
+ | ---
+ | ...
+
+-- Emulate a situation when next_leader wins the elections. It can't do 
that in
+-- this configuration, obviously, because it's behind the 'other' node, 
so set
+-- quorum to 1 and imagine there are 2 more servers which would vote for
+-- next_leader.
+-- Also, make the instance ignore synchronization with other replicas.
+-- Otherwise it would stall for replication_sync_timeout. This is due 
to the
+-- nature of the test and may be ignored (we restart the instance to 
simulate
+-- a situation when some rows from the old leader were not received).
+test_run:cmd('start server '..next_leader..' with args="1 0.4 candidate 
1"')
+ | ---
+ | - true
+ | ...
+assert(get_leader(is_possible_leader) == next_leader_nr)
+ | ---
+ | - true
+ | ...
+test_run:switch(other)
+ | ---
+ | - true
+ | ...
+-- New leader didn't know about the unconfirmed rows but still rolled 
them back.
+test_run:wait_cond(function() return box.space.test:get{2} == nil end)
+ | ---
+ | - true
+ | ...
+
+test_run:switch('default')
+ | ---
+ | - true
+ | ...
+test_run:switch(next_leader)
+ | ---
+ | - true
+ | ...
+box.space.test:select{} -- 1
+ | ---
+ | - - [1]
+ | ...
+box.cfg{election_mode='voter'}
+ | ---
+ | ...
+-- Old leader returns and old unconfirmed rows from it must be ignored.
+test_run:switch('default')
+ | ---
+ | - true
+ | ...
+-- Make old leader win the elections.
+test_run:cmd('start server '..leader..' with args="1 0.4 candidate 1"')
+ | ---
+ | - true
+ | ...
+is_possible_leader[leader_nr] = true
+ | ---
+ | ...
+assert(get_leader(is_possible_leader) == leader_nr)
+ | ---
+ | - true
+ | ...
+test_run:switch(next_leader)
+ | ---
+ | - true
+ | ...
+box.space.test:select{} -- 1
+ | ---
+ | - - [1]
+ | ...
+test_run:wait_upstream(1, {status='follow'})
+ | ---
+ | - true
+ | ...
+
+-- Cleanup.
+test_run:switch('default')
+ | ---
+ | - true
+ | ...
+test_run:drop_cluster(SERVERS)
+ | ---
+ | ...
diff --git a/test/replication/gh-5445-leader-incosistency.test.lua 
b/test/replication/gh-5445-leader-incosistency.test.lua
new file mode 100644
index 000000000..94beea966
--- /dev/null
+++ b/test/replication/gh-5445-leader-incosistency.test.lua
@@ -0,0 +1,108 @@
+test_run = require("test_run").new()
+
+is_leader_cmd = "return box.info.election.state == 'leader'"
+
+-- Auxiliary.
+test_run:cmd('setopt delimiter ";"')
+function get_leader(nrs)
+    local leader_nr = 0
+    test_run:wait_cond(function()
+        for nr, do_check in pairs(nrs) do
+            if do_check then
+                local is_leader = test_run:eval('election_replica'..nr,
+                                                is_leader_cmd)[1]
+                if is_leader then
+                    leader_nr = nr
+                    return true
+                end
+            end
+        end
+        return false
+    end)
+    assert(leader_nr ~= 0)
+    return leader_nr
+end;
+
+function name(id)
+    return 'election_replica'..id
+end;
+test_run:cmd('setopt delimiter ""');
+
+--
+-- gh-5445: make sure rolled back rows do not reappear once old leader 
returns
+-- to cluster.
+--
+SERVERS = {'election_replica1', 'election_replica2' ,'election_replica3'}
+test_run:create_cluster(SERVERS, "replication", {args='2 0.4'})
+test_run:wait_fullmesh(SERVERS)
+
+-- Any of the three instances may bootstrap the cluster and become leader.
+is_possible_leader = {true, true, true}
+leader_nr = get_leader(is_possible_leader)
+leader = name(leader_nr)
+next_leader_nr = ((leader_nr - 1) % 3 + 1) % 3 + 1 -- {1, 2, 3} -> {2, 
3, 1}
+next_leader = name(next_leader_nr)
+other_nr = ((leader_nr - 1) % 3 + 2) % 3 + 1 -- {1, 2, 3} -> {3, 1, 2}
+other = name(other_nr)
+
+test_run:switch(leader)
+box.ctl.wait_rw()
+_ = box.schema.space.create('test', {is_sync=true})
+_ = box.space.test:create_index('pk')
+box.space.test:insert{1}
+
+-- Simulate a situation when the instance which will become the next leader
+-- doesn't know of unconfirmed rows. It should roll them back anyways 
and do not
+-- accept them once they actually appear from the old leader.
+-- So, stop the instance which'll be the next leader.
+test_run:switch('default')
+test_run:cmd('stop server '..next_leader)
+test_run:switch(leader)
+-- Insert some unconfirmed data.
+box.cfg{replication_synchro_quorum=3, replication_synchro_timeout=1000}
+fib = require('fiber').create(box.space.test.insert, box.space.test, {2})
+fib:status()
+
+-- 'other', 'leader', 'next_leader' are defined on 'default' node, 
hence the
+-- double switches.
+test_run:switch('default')
+test_run:switch(other)
+-- Wait until the rows are replicated to the other instance.
+test_run:wait_cond(function() return box.space.test:get{2} ~= nil end)
+box.cfg{election_mode='voter'}
+-- Old leader is gone.
+test_run:switch('default')
+test_run:cmd('stop server '..leader)
+is_possible_leader[leader_nr] = false
+
+-- Emulate a situation when next_leader wins the elections. It can't do 
that in
+-- this configuration, obviously, because it's behind the 'other' node, 
so set
+-- quorum to 1 and imagine there are 2 more servers which would vote for
+-- next_leader.
+-- Also, make the instance ignore synchronization with other replicas.
+-- Otherwise it would stall for replication_sync_timeout. This is due 
to the
+-- nature of the test and may be ignored (we restart the instance to 
simulate
+-- a situation when some rows from the old leader were not received).
+test_run:cmd('start server '..next_leader..' with args="1 0.4 candidate 
1"')
+assert(get_leader(is_possible_leader) == next_leader_nr)
+test_run:switch(other)
+-- New leader didn't know about the unconfirmed rows but still rolled 
them back.
+test_run:wait_cond(function() return box.space.test:get{2} == nil end)
+
+test_run:switch('default')
+test_run:switch(next_leader)
+box.space.test:select{} -- 1
+box.cfg{election_mode='voter'}
+-- Old leader returns and old unconfirmed rows from it must be ignored.
+test_run:switch('default')
+-- Make old leader win the elections.
+test_run:cmd('start server '..leader..' with args="1 0.4 candidate 1"')
+is_possible_leader[leader_nr] = true
+assert(get_leader(is_possible_leader) == leader_nr)
+test_run:switch(next_leader)
+box.space.test:select{} -- 1
+test_run:wait_upstream(1, {status='follow'})
+
+-- Cleanup.
+test_run:switch('default')
+test_run:drop_cluster(SERVERS)
diff --git a/test/replication/suite.cfg b/test/replication/suite.cfg
index aff5fda26..0a270d3d6 100644
--- a/test/replication/suite.cfg
+++ b/test/replication/suite.cfg
@@ -17,6 +17,7 @@
      "gh-4424-misc-orphan-on-reconfiguration-error.test.lua": {},
      "gh-5426-election-on-off.test.lua": {},
      "gh-5433-election-restart-recovery.test.lua": {},
+    "gh-5445-leader-incosistency.test.lua": {},
      "gh-5506-election-on-off.test.lua": {},
      "once.test.lua": {},
      "on_replace.test.lua": {},

-- 
Serge Petrenko