Commit 3e011ae2 authored by Willem Thiart's avatar Willem Thiart
Browse files

Fixes #37

Avoid deleting logs because of out-of-order/quick heartbeats.
parent 67ead6cb
...@@ -98,6 +98,8 @@ void raft_delete_entry_from_idx(raft_server_t* me_, int idx) ...@@ -98,6 +98,8 @@ void raft_delete_entry_from_idx(raft_server_t* me_, int idx)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
assert(me->commit_idx < idx);
if (idx <= me->voting_cfg_change_log_idx) if (idx <= me->voting_cfg_change_log_idx)
me->voting_cfg_change_log_idx = -1; me->voting_cfg_change_log_idx = -1;
...@@ -371,7 +373,6 @@ int raft_recv_appendentries( ...@@ -371,7 +373,6 @@ int raft_recv_appendentries(
{ {
__log(me_, node, "AE term doesn't match prev_term (ie. %d vs %d) ci:%d pli:%d", __log(me_, node, "AE term doesn't match prev_term (ie. %d vs %d) ci:%d pli:%d",
e->term, ae->prev_log_term, raft_get_current_idx(me_), ae->prev_log_idx); e->term, ae->prev_log_term, raft_get_current_idx(me_), ae->prev_log_idx);
assert(me->commit_idx < ae->prev_log_idx);
/* Delete all the following log entries because they don't match */ /* Delete all the following log entries because they don't match */
raft_delete_entry_from_idx(me_, ae->prev_log_idx); raft_delete_entry_from_idx(me_, ae->prev_log_idx);
r->current_idx = ae->prev_log_idx - 1; r->current_idx = ae->prev_log_idx - 1;
...@@ -382,9 +383,13 @@ int raft_recv_appendentries( ...@@ -382,9 +383,13 @@ int raft_recv_appendentries(
/* 3. If an existing entry conflicts with a new one (same index /* 3. If an existing entry conflicts with a new one (same index
but different terms), delete the existing entry and all that but different terms), delete the existing entry and all that
follow it (§5.3) */ follow it (§5.3) */
if (ae->n_entries == 0 && 0 < ae->prev_log_idx && ae->prev_log_idx + 1 < raft_get_current_idx(me_)) if (0 < ae->prev_log_idx && ae->prev_log_idx + 1 < raft_get_current_idx(me_))
{ {
assert(me->commit_idx < ae->prev_log_idx + 1); /* Heartbeats shouldn't cause logs to be deleted. Heartbeats might be
* sent before the leader received the last appendentries response */
if (ae->n_entries != 0 &&
/* this is an old out-of-order appendentry message */
me->commit_idx < ae->prev_log_idx + 1)
raft_delete_entry_from_idx(me_, ae->prev_log_idx + 1); raft_delete_entry_from_idx(me_, ae->prev_log_idx + 1);
} }
...@@ -397,9 +402,8 @@ int raft_recv_appendentries( ...@@ -397,9 +402,8 @@ int raft_recv_appendentries(
int ety_index = ae->prev_log_idx + 1 + i; int ety_index = ae->prev_log_idx + 1 + i;
raft_entry_t* existing_ety = raft_get_entry_from_idx(me_, ety_index); raft_entry_t* existing_ety = raft_get_entry_from_idx(me_, ety_index);
r->current_idx = ety_index; r->current_idx = ety_index;
if (existing_ety && existing_ety->term != ety->term) if (existing_ety && existing_ety->term != ety->term && me->commit_idx < ety_index)
{ {
assert(me->commit_idx < ety_index);
raft_delete_entry_from_idx(me_, ety_index); raft_delete_entry_from_idx(me_, ety_index);
break; break;
} }
......
...@@ -1349,17 +1349,21 @@ void TestRaft_follower_recv_appendentries_delete_entries_if_current_idx_greater_ ...@@ -1349,17 +1349,21 @@ void TestRaft_follower_recv_appendentries_delete_entries_if_current_idx_greater_
raft_entry_t *ety_appended; raft_entry_t *ety_appended;
__create_mock_entries_for_conflict_tests(tc, r, strs); __create_mock_entries_for_conflict_tests(tc, r, strs);
CuAssertIntEquals(tc, 3, raft_get_log_count(r));
memset(&ae, 0, sizeof(msg_appendentries_t)); memset(&ae, 0, sizeof(msg_appendentries_t));
ae.term = 2; ae.term = 2;
ae.prev_log_idx = 1; ae.prev_log_idx = 1;
ae.prev_log_term = 1; ae.prev_log_term = 1;
ae.entries = NULL; msg_entry_t e[1];
ae.n_entries = 0; memset(&e, 0, sizeof(msg_entry_t) * 1);
e[0].id = 1;
ae.entries = e;
ae.n_entries = 1;
raft_recv_appendentries(r, raft_get_node(r, 2), &ae, &aer); raft_recv_appendentries(r, raft_get_node(r, 2), &ae, &aer);
CuAssertTrue(tc, 1 == aer.success); CuAssertTrue(tc, 1 == aer.success);
CuAssertTrue(tc, 1 == raft_get_log_count(r)); CuAssertIntEquals(tc, 2, raft_get_log_count(r));
CuAssertTrue(tc, NULL != (ety_appended = raft_get_entry_from_idx(r, 1))); CuAssertTrue(tc, NULL != (ety_appended = raft_get_entry_from_idx(r, 1)));
CuAssertTrue(tc, !strncmp(ety_appended->data.buf, strs[0], 3)); CuAssertTrue(tc, !strncmp(ety_appended->data.buf, strs[0], 3));
} }
...@@ -1667,6 +1671,169 @@ void TestRaft_follower_dont_grant_vote_if_candidate_has_a_less_complete_log( ...@@ -1667,6 +1671,169 @@ void TestRaft_follower_dont_grant_vote_if_candidate_has_a_less_complete_log(
CuAssertIntEquals(tc, 1, rvr.vote_granted); CuAssertIntEquals(tc, 1, rvr.vote_granted);
} }
void TestRaft_follower_recv_appendentries_heartbeat_does_not_overwrite_logs(
CuTest * tc)
{
raft_cbs_t funcs = {
.persist_term = __raft_persist_term,
};
void *r = raft_new();
raft_set_callbacks(r, &funcs, NULL);
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0);
msg_appendentries_t ae;
msg_appendentries_response_t aer;
memset(&ae, 0, sizeof(msg_appendentries_t));
ae.term = 1;
ae.prev_log_idx = 0;
ae.prev_log_term = 1;
/* include entries */
msg_entry_t e[4];
memset(&e, 0, sizeof(msg_entry_t) * 4);
e[0].term = 1;
e[0].id = 1;
ae.entries = e;
ae.n_entries = 1;
raft_recv_appendentries(r, raft_get_node(r, 2), &ae, &aer);
/* The server sends a follow up AE
* NOTE: the server has received a response from the last AE so
* prev_log_idx has been incremented */
memset(&ae, 0, sizeof(msg_appendentries_t));
ae.term = 1;
ae.prev_log_idx = 1;
ae.prev_log_term = 1;
/* include entries */
memset(&e, 0, sizeof(msg_entry_t) * 4);
e[0].term = 1;
e[0].id = 2;
e[1].term = 1;
e[1].id = 3;
e[2].term = 1;
e[2].id = 4;
e[3].term = 1;
e[3].id = 5;
ae.entries = e;
ae.n_entries = 4;
raft_recv_appendentries(r, raft_get_node(r, 2), &ae, &aer);
/* receive a heartbeat
* NOTE: the leader hasn't received the response to the last AE so it can
* only assume prev_Log_idx is still 1 */
memset(&ae, 0, sizeof(msg_appendentries_t));
ae.term = 1;
ae.prev_log_term = 1;
ae.prev_log_idx = 1;
/* receipt of appendentries changes commit idx */
raft_recv_appendentries(r, raft_get_node(r, 2), &ae, &aer);
CuAssertTrue(tc, 1 == aer.success);
CuAssertIntEquals(tc, 5, raft_get_current_idx(r));
}
void TestRaft_follower_recv_appendentries_does_not_deleted_commited_entries(
CuTest * tc)
{
raft_cbs_t funcs = {
.persist_term = __raft_persist_term,
};
void *r = raft_new();
raft_set_callbacks(r, &funcs, NULL);
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0);
msg_appendentries_t ae;
msg_appendentries_response_t aer;
memset(&ae, 0, sizeof(msg_appendentries_t));
ae.term = 1;
ae.prev_log_idx = 0;
ae.prev_log_term = 1;
/* include entries */
msg_entry_t e[5];
memset(&e, 0, sizeof(msg_entry_t) * 4);
e[0].term = 1;
e[0].id = 1;
ae.entries = e;
ae.n_entries = 1;
raft_recv_appendentries(r, raft_get_node(r, 2), &ae, &aer);
/* Follow up AE. Node responded with success */
memset(&ae, 0, sizeof(msg_appendentries_t));
ae.term = 1;
ae.prev_log_idx = 1;
ae.prev_log_term = 1;
/* include entries */
memset(&e, 0, sizeof(msg_entry_t) * 4);
e[0].term = 1;
e[0].id = 2;
e[1].term = 1;
e[1].id = 3;
e[2].term = 1;
e[2].id = 4;
e[3].term = 1;
e[3].id = 5;
ae.entries = e;
ae.n_entries = 4;
ae.leader_commit = 4;
raft_recv_appendentries(r, raft_get_node(r, 2), &ae, &aer);
/* The server sends a follow up AE */
memset(&ae, 0, sizeof(msg_appendentries_t));
ae.term = 1;
ae.prev_log_idx = 1;
ae.prev_log_term = 1;
/* include entries */
memset(&e, 0, sizeof(msg_entry_t) * 5);
e[0].term = 1;
e[0].id = 2;
e[1].term = 1;
e[1].id = 3;
e[2].term = 1;
e[2].id = 4;
e[3].term = 1;
e[3].id = 5;
e[4].term = 1;
e[4].id = 6;
ae.entries = e;
ae.n_entries = 5;
ae.leader_commit = 4;
raft_recv_appendentries(r, raft_get_node(r, 2), &ae, &aer);
CuAssertTrue(tc, 1 == aer.success);
CuAssertIntEquals(tc, 6, raft_get_current_idx(r));
CuAssertIntEquals(tc, 4, raft_get_commit_idx(r));
/* The server sends a follow up AE.
* This appendentry forces the node to check if it's going to delete
* commited logs */
memset(&ae, 0, sizeof(msg_appendentries_t));
ae.term = 1;
ae.prev_log_idx = 3;
ae.prev_log_term = 1;
/* include entries */
memset(&e, 0, sizeof(msg_entry_t) * 5);
e[0].id = 1;
e[0].id = 5;
e[1].term = 1;
e[1].id = 6;
e[2].term = 1;
e[2].id = 7;
ae.entries = e;
ae.n_entries = 3;
ae.leader_commit = 4;
raft_recv_appendentries(r, raft_get_node(r, 2), &ae, &aer);
CuAssertTrue(tc, 1 == aer.success);
CuAssertIntEquals(tc, 6, raft_get_current_idx(r));
}
void TestRaft_candidate_becomes_candidate_is_candidate(CuTest * tc) void TestRaft_candidate_becomes_candidate_is_candidate(CuTest * tc)
{ {
raft_cbs_t funcs = { raft_cbs_t funcs = {
......
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