Commit 700eebe6 authored by Willem Thiart's avatar Willem Thiart
Browse files

Remove RAFT_LOGTYPE_SNAPSHOT

The RAFT_LOGTYPE_SNAPSHOT was a bad design choice. It's better to not
have this marker entry so that users don't have to check for the log
entry type in the callbacks. Also, loading a snapshot correctly created
this marker entry, but the compaction side did not. It's better to
remove this log type than to fix this asymmetry.

Fixes the following edge cases:
- Server crashes if the last snapshot index is N and a appendentries
message pops entries all the way back to index N.
- Server sends an appendentries message with an invalid prev_log_idx and
prev_log_term, causing the peer to crash.
parent 9de8af47
......@@ -440,7 +440,7 @@ The process works like this:
2. Save the current membership details to the snapshot.
3. Save the finite state machine to the snapshot.
4. End snapshotting with ``raft_end_snapshot``.
5. When the ``send_snapshot`` callback fires, the user must propogate the snapshot to the other node.
5. When the ``send_snapshot`` callback fires, the user must propogate the snapshot to the peer.
6. Once the peer has the snapshot, they call ``raft_begin_load_snapshot``.
7. Peer calls ``raft_add_node`` to add nodes as per the snapshot's membership info.
8. Peer calls ``raft_node_set_voting`` to nodes as per the snapshot's membership info.
......
......@@ -62,7 +62,6 @@ typedef enum {
* Removing nodes is a 2 step process: first demote, then remove.
*/
RAFT_LOGTYPE_REMOVE_NODE,
RAFT_LOGTYPE_SNAPSHOT,
/**
* Users can piggyback the entry mechanism by specifying log types that
* are higher than RAFT_LOGTYPE_NUM.
......
......@@ -80,21 +80,7 @@ int log_load_from_snapshot(log_t *me_, int idx, int term)
log_private_t* me = (log_private_t*)me_;
log_clear(me_);
raft_entry_t ety;
ety.data.len = 0;
ety.id = 1;
ety.term = term;
ety.type = RAFT_LOGTYPE_SNAPSHOT;
int e = log_append_entry(me_, &ety);
if (e != 0)
{
assert(0);
return e;
}
me->base = idx - 1;
me->base = idx;
return 0;
}
......@@ -202,10 +188,7 @@ raft_entry_t* log_get_at_idx(log_t* me_, int idx)
if (idx == 0)
return NULL;
if (idx <= me->base)
return NULL;
if (me->base + me->count < idx)
if (me->base + me->count < idx || idx <= me->base)
return NULL;
/* idx starts at 1 */
......
......@@ -125,7 +125,6 @@ void raft_clear(raft_server_t* me_)
me->last_applied_idx = 0;
me->num_nodes = 0;
me->node = NULL;
me->voting_cfg_change_log_idx = 0;
log_clear(me->log);
}
......@@ -431,15 +430,25 @@ int raft_recv_appendentries(
{
raft_entry_t* ety = raft_get_entry_from_idx(me_, ae->prev_log_idx);
/* Is a snapshot */
if (ae->prev_log_idx == me->snapshot_last_idx)
{
if (me->snapshot_last_term != ae->prev_log_term)
{
/* Should never happen; something is seriously wrong! */
__log(me_, node, "Snapshot AE prev conflicts with committed entry");
e = RAFT_ERR_SHUTDOWN;
goto out;
}
}
/* 2. Reply false if log doesn't contain an entry at prevLogIndex
whose term matches prevLogTerm (§5.3) */
if (!ety)
else if (!ety)
{
__log(me_, node, "AE no log at prev_idx %d", ae->prev_log_idx);
goto out;
}
if (ety->term != ae->prev_log_term)
else if (ety->term != ae->prev_log_term)
{
__log(me_, node, "AE term doesn't match prev_term (ie. %d vs %d) ci:%d comi:%d lcomi:%d pli:%d",
ety->term, ae->prev_log_term, raft_get_current_idx(me_),
......@@ -546,10 +555,20 @@ static int __should_grant_vote(raft_server_private_t* me, msg_requestvote_t* vr)
return 1;
raft_entry_t* ety = raft_get_entry_from_idx((void*)me, current_idx);
if (ety->term < vr->last_log_term)
int ety_term;
// TODO: add test
if (ety)
ety_term = ety->term;
else if (!ety && me->snapshot_last_idx == current_idx)
ety_term = me->snapshot_last_term;
else
return 0;
if (ety_term < vr->last_log_term)
return 1;
if (vr->last_log_term == ety->term && current_idx <= vr->last_log_idx)
if (vr->last_log_term == ety_term && current_idx <= vr->last_log_idx)
return 1;
return 0;
......@@ -589,7 +608,7 @@ int raft_recv_requestvote(raft_server_t* me_,
else
r->vote_granted = 0;
/* there must be in an election. */
/* must be in an election. */
me->current_leader = NULL;
me->timeout_elapsed = 0;
......@@ -887,11 +906,16 @@ int raft_send_appendentries(raft_server_t* me_, raft_node_t* node)
if (1 < next_idx)
{
raft_entry_t* prev_ety = raft_get_entry_from_idx(me_, next_idx - 1);
ae.prev_log_idx = next_idx - 1;
if (prev_ety)
ae.prev_log_term = prev_ety->term;
else
if (!prev_ety)
{
ae.prev_log_idx = me->snapshot_last_idx;
ae.prev_log_term = me->snapshot_last_term;
}
else
{
ae.prev_log_idx = next_idx - 1;
ae.prev_log_term = prev_ety->term;
}
}
__log(me_, node, "sending appendentries node: ci:%d comi:%d t:%d lc:%d pli:%d plt:%d",
......@@ -1247,9 +1271,12 @@ int raft_end_snapshot(raft_server_t *me_)
if (!me->snapshot_in_progress || me->snapshot_last_idx == 0)
return -1;
assert(raft_get_num_snapshottable_logs(me_) != 0);
assert(me->snapshot_last_idx == raft_get_commit_idx(me_));
/* If needed, remove compacted logs */
int i = raft_get_first_entry_idx(me_), end = raft_get_commit_idx(me_);
for (; i <= end; i++)
int i = 0, end = raft_get_num_snapshottable_logs(me_);
for (; i < end; i++)
{
raft_entry_t* _ety;
int e = raft_poll_entry(me_, &_ety);
......
......@@ -399,10 +399,7 @@ void TestLog_load_from_snapshot(CuTest * tc)
CuAssertIntEquals(tc, 0, log_get_current_idx(l));
CuAssertIntEquals(tc, 0, log_load_from_snapshot(l, 10, 5));
CuAssertIntEquals(tc, 10, log_get_current_idx(l));
/* this is just a marker
* it should never be sent to any nodes because it is part of a snapshot */
CuAssertIntEquals(tc, 1, log_count(l));
CuAssertIntEquals(tc, 0, log_count(l));
}
void TestLog_load_from_snapshot_clears_log(CuTest * tc)
......@@ -422,7 +419,7 @@ void TestLog_load_from_snapshot_clears_log(CuTest * tc)
CuAssertIntEquals(tc, 2, log_get_current_idx(l));
CuAssertIntEquals(tc, 0, log_load_from_snapshot(l, 10, 5));
CuAssertIntEquals(tc, 1, log_count(l));
CuAssertIntEquals(tc, 0, log_count(l));
CuAssertIntEquals(tc, 10, log_get_current_idx(l));
}
......
......@@ -362,7 +362,7 @@ void TestRaft_follower_load_from_snapshot(CuTest * tc)
CuAssertIntEquals(tc, 0, raft_get_log_count(r));
CuAssertIntEquals(tc, 0, raft_begin_load_snapshot(r, 5, 5));
CuAssertIntEquals(tc, 0, raft_end_load_snapshot(r));
CuAssertIntEquals(tc, 1, raft_get_log_count(r));
CuAssertIntEquals(tc, 0, raft_get_log_count(r));
CuAssertIntEquals(tc, 0, raft_get_num_snapshottable_logs(r));
CuAssertIntEquals(tc, 5, raft_get_commit_idx(r));
CuAssertIntEquals(tc, 5, raft_get_last_applied_idx(r));
......@@ -418,7 +418,7 @@ void TestRaft_follower_load_from_snapshot_fails_if_already_loaded(CuTest * tc)
CuAssertIntEquals(tc, 0, raft_get_log_count(r));
CuAssertIntEquals(tc, 0, raft_begin_load_snapshot(r, 5, 5));
CuAssertIntEquals(tc, 0, raft_end_load_snapshot(r));
CuAssertIntEquals(tc, 1, raft_get_log_count(r));
CuAssertIntEquals(tc, 0, raft_get_log_count(r));
CuAssertIntEquals(tc, 0, raft_get_num_snapshottable_logs(r));
CuAssertIntEquals(tc, 5, raft_get_commit_idx(r));
CuAssertIntEquals(tc, 5, raft_get_last_applied_idx(r));
......@@ -537,7 +537,7 @@ void TestRaft_leader_sends_appendentries_when_node_next_index_was_compacted(CuTe
raft_set_current_term(r, 2);
CuAssertIntEquals(tc, 0, raft_send_appendentries(r, node));
CuAssertIntEquals(tc, 2, ae.term);
CuAssertIntEquals(tc, 2, ae.prev_log_idx);
CuAssertIntEquals(tc, 3, ae.prev_log_idx);
CuAssertIntEquals(tc, 2, ae.prev_log_term);
}
......@@ -578,3 +578,69 @@ void TestRaft_recv_entry_fails_if_snapshot_in_progress(CuTest* tc)
ety.type = RAFT_LOGTYPE_ADD_NODE;
CuAssertIntEquals(tc, RAFT_ERR_SNAPSHOT_IN_PROGRESS, raft_recv_entry(r, &ety, &cr));
}
void TestRaft_follower_recv_appendentries_is_successful_when_previous_log_idx_equals_snapshot_last_idx(
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);
CuAssertIntEquals(tc, 0, raft_begin_load_snapshot(r, 2, 2));
CuAssertIntEquals(tc, 0, raft_end_load_snapshot(r));
msg_appendentries_t ae;
msg_appendentries_response_t aer;
memset(&ae, 0, sizeof(msg_appendentries_t));
ae.term = 3;
ae.prev_log_idx = 2;
ae.prev_log_term = 2;
/* include entries */
msg_entry_t e[5];
memset(&e, 0, sizeof(msg_entry_t) * 4);
e[0].term = 3;
e[0].id = 3;
ae.entries = e;
ae.n_entries = 1;
CuAssertIntEquals(tc, 0, raft_recv_appendentries(r, raft_get_node(r, 2), &ae, &aer));
CuAssertIntEquals(tc, 1, aer.success);
}
void TestRaft_leader_sends_appendentries_with_correct_prev_log_idx_when_snapshotted(
CuTest * tc)
{
raft_cbs_t funcs = {
.send_appendentries = sender_appendentries,
.log = NULL
};
void *sender = sender_new(NULL);
void *r = raft_new();
raft_set_callbacks(r, &funcs, sender);
CuAssertTrue(tc, NULL != raft_add_node(r, NULL, 1, 1));
CuAssertTrue(tc, NULL != raft_add_node(r, NULL, 2, 0));
CuAssertIntEquals(tc, 0, raft_begin_load_snapshot(r, 2, 4));
CuAssertIntEquals(tc, 0, raft_end_load_snapshot(r));
/* i'm leader */
raft_set_state(r, RAFT_STATE_LEADER);
raft_node_t* p = raft_get_node_from_idx(r, 1);
CuAssertTrue(tc, NULL != p);
raft_node_set_next_idx(p, 4);
/* receive appendentries messages */
raft_send_appendentries(r, p);
msg_appendentries_t* ae = sender_poll_msg_data(sender);
CuAssertTrue(tc, NULL != ae);
CuAssertIntEquals(tc, 2, ae->prev_log_term);
CuAssertIntEquals(tc, 4, ae->prev_log_idx);
}
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