Commit 29bf1bcf authored by Willem Thiart's avatar Willem Thiart
Browse files

raft_recv_entry won't accept 1 < voting changes

parent 8a7b07a0
...@@ -180,6 +180,7 @@ typedef int ( ...@@ -180,6 +180,7 @@ typedef int (
); );
/** Callback for detecting when non-voting nodes have obtained enough logs. /** Callback for detecting when non-voting nodes have obtained enough logs.
* This triggers only when there are no pending configuration changes.
* @param[in] raft The Raft server making this callback * @param[in] raft The Raft server making this callback
* @param[in] user_data User data that is passed from Raft server * @param[in] user_data User data that is passed from Raft server
* @param[in] node The node */ * @param[in] node The node */
...@@ -635,4 +636,14 @@ void raft_apply_all(raft_server_t* me_); ...@@ -635,4 +636,14 @@ void raft_apply_all(raft_server_t* me_);
* losing it's consensus guarantees. */ * losing it's consensus guarantees. */
void raft_become_leader(raft_server_t* me); void raft_become_leader(raft_server_t* me);
/** Determine if entry is voting configuration change.
* @param[in] ety The entry to query.
* @return 1 if this is a voting configuration change. */
int raft_entry_is_voting_cfg_change(raft_entry_t* ety);
/** Determine if entry is configuration change.
* @param[in] ety The entry to query.
* @return 1 if this is a configuration change. */
int raft_entry_is_cfg_change(raft_entry_t* ety);
#endif /* RAFT_H_ */ #endif /* RAFT_H_ */
...@@ -56,6 +56,9 @@ typedef struct { ...@@ -56,6 +56,9 @@ typedef struct {
/* my node ID */ /* my node ID */
raft_node_t* node; raft_node_t* node;
/* the log which has a voting cfg change, otherwise -1 */
int voting_cfg_change_log_idx;
} raft_server_private_t; } raft_server_private_t;
void raft_election_start(raft_server_t* me); void raft_election_start(raft_server_t* me);
......
...@@ -49,6 +49,7 @@ raft_server_t* raft_new() ...@@ -49,6 +49,7 @@ raft_server_t* raft_new()
me->request_timeout = 200; me->request_timeout = 200;
me->election_timeout = 1000; me->election_timeout = 1000;
me->log = log_new(); me->log = log_new();
me->voting_cfg_change_log_idx = -1;
raft_set_state((raft_server_t*)me, RAFT_STATE_FOLLOWER); raft_set_state((raft_server_t*)me, RAFT_STATE_FOLLOWER);
me->current_leader = NULL; me->current_leader = NULL;
return (raft_server_t*)me; return (raft_server_t*)me;
...@@ -220,6 +221,7 @@ int raft_recv_appendentries_response(raft_server_t* me_, ...@@ -220,6 +221,7 @@ int raft_recv_appendentries_response(raft_server_t* me_,
raft_node_set_match_idx(node, r->current_idx); raft_node_set_match_idx(node, r->current_idx);
if (!raft_node_is_voting(node) && if (!raft_node_is_voting(node) &&
-1 == me->voting_cfg_change_log_idx &&
raft_get_current_idx(me_) <= r->current_idx + 1 && raft_get_current_idx(me_) <= r->current_idx + 1 &&
me->cb.node_has_sufficient_logs && me->cb.node_has_sufficient_logs &&
0 == raft_node_has_sufficient_logs(node) 0 == raft_node_has_sufficient_logs(node)
...@@ -513,6 +515,11 @@ int raft_recv_entry(raft_server_t* me_, ...@@ -513,6 +515,11 @@ int raft_recv_entry(raft_server_t* me_,
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
int i; int i;
/* Only one voting cfg change at a time */
if (raft_entry_is_voting_cfg_change(e))
if (-1 != me->voting_cfg_change_log_idx)
return -1;
if (!raft_is_leader(me_)) if (!raft_is_leader(me_))
return -1; return -1;
...@@ -546,6 +553,10 @@ int raft_recv_entry(raft_server_t* me_, ...@@ -546,6 +553,10 @@ int raft_recv_entry(raft_server_t* me_,
r->id = e->id; r->id = e->id;
r->idx = raft_get_current_idx(me_); r->idx = raft_get_current_idx(me_);
r->term = me->current_term; r->term = me->current_term;
if (raft_entry_is_voting_cfg_change(e))
me->voting_cfg_change_log_idx = raft_get_current_idx(me_);
return 0; return 0;
} }
...@@ -571,6 +582,10 @@ int raft_send_requestvote(raft_server_t* me_, raft_node_t* node) ...@@ -571,6 +582,10 @@ int raft_send_requestvote(raft_server_t* me_, raft_node_t* node)
int raft_append_entry(raft_server_t* me_, raft_entry_t* ety) int raft_append_entry(raft_server_t* me_, raft_entry_t* ety)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
if (raft_entry_is_voting_cfg_change(ety))
me->voting_cfg_change_log_idx = raft_get_current_idx(me_);
return log_append_entry(me->log, ety); return log_append_entry(me->log, ety);
} }
...@@ -594,6 +609,11 @@ int raft_apply_entry(raft_server_t* me_) ...@@ -594,6 +609,11 @@ int raft_apply_entry(raft_server_t* me_)
me->last_applied_idx++; me->last_applied_idx++;
if (me->cb.applylog) if (me->cb.applylog)
me->cb.applylog(me_, me->udata, e); me->cb.applylog(me_, me->udata, e);
/* voting cfg change is now complete */
if (log_idx == me->voting_cfg_change_log_idx)
me->voting_cfg_change_log_idx = -1;
return 0; return 0;
} }
...@@ -660,7 +680,7 @@ void raft_send_appendentries_all(raft_server_t* me_) ...@@ -660,7 +680,7 @@ void raft_send_appendentries_all(raft_server_t* me_)
me->timeout_elapsed = 0; me->timeout_elapsed = 0;
for (i = 0; i < me->num_nodes; i++) for (i = 0; i < me->num_nodes; i++)
if (me->node != me->nodes[i] && raft_node_is_voting(me->nodes[i])) if (me->node != me->nodes[i])
raft_send_appendentries(me_, me->nodes[i]); raft_send_appendentries(me_, me->nodes[i]);
} }
...@@ -764,3 +784,17 @@ void raft_apply_all(raft_server_t* me_) ...@@ -764,3 +784,17 @@ void raft_apply_all(raft_server_t* me_)
while (raft_get_last_applied_idx(me_) < raft_get_commit_idx(me_)) while (raft_get_last_applied_idx(me_) < raft_get_commit_idx(me_))
raft_apply_entry(me_); raft_apply_entry(me_);
} }
int raft_entry_is_voting_cfg_change(raft_entry_t* ety)
{
return RAFT_LOGTYPE_ADD_NODE == ety->type ||
RAFT_LOGTYPE_REMOVE_NODE == ety->type;
}
int raft_entry_is_cfg_change(raft_entry_t* ety)
{
return (
RAFT_LOGTYPE_ADD_NODE == ety->type ||
RAFT_LOGTYPE_ADD_NONVOTING_NODE == ety->type ||
RAFT_LOGTYPE_REMOVE_NODE == ety->type);
}
...@@ -384,6 +384,31 @@ void TestRaft_server_recv_entry_auto_commits_if_we_are_the_only_node(CuTest * tc ...@@ -384,6 +384,31 @@ void TestRaft_server_recv_entry_auto_commits_if_we_are_the_only_node(CuTest * tc
CuAssertTrue(tc, 1 == raft_get_commit_idx(r)); CuAssertTrue(tc, 1 == raft_get_commit_idx(r));
} }
void TestRaft_server_recv_entry_fails_if_there_is_already_a_voting_change(CuTest * tc)
{
void *r = raft_new();
raft_add_node(r, NULL, 1, 1);
raft_set_election_timeout(r, 1000);
raft_become_leader(r);
CuAssertTrue(tc, 0 == raft_get_commit_idx(r));
/* entry message */
msg_entry_t ety;
ety.type = RAFT_LOGTYPE_ADD_NODE;
ety.id = 1;
ety.data.buf = "entry";
ety.data.len = strlen("entry");
/* receive entry */
msg_entry_response_t cr;
CuAssertTrue(tc, 0 == raft_recv_entry(r, &ety, &cr));
CuAssertTrue(tc, 1 == raft_get_log_count(r));
ety.id = 2;
CuAssertTrue(tc, -1 == raft_recv_entry(r, &ety, &cr));
CuAssertTrue(tc, 1 == raft_get_commit_idx(r));
}
void TestRaft_server_cfg_sets_num_nodes(CuTest * tc) void TestRaft_server_cfg_sets_num_nodes(CuTest * tc)
{ {
void *r = raft_new(); void *r = raft_new();
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment