Unverified Commit d8799dd9 authored by Ozan Tezcan's avatar Ozan Tezcan Committed by GitHub
Browse files

Fix : send snapshot when node's next index is snapshot's last included index (#47)

parent 2896eb00
...@@ -1219,41 +1219,41 @@ raft_entry_t** raft_get_entries_from_idx(raft_server_t* me_, raft_index_t idx, i ...@@ -1219,41 +1219,41 @@ raft_entry_t** raft_get_entries_from_idx(raft_server_t* me_, raft_index_t idx, i
int raft_send_appendentries(raft_server_t* me_, raft_node_t* node) int raft_send_appendentries(raft_server_t* me_, raft_node_t* node)
{ {
if (!raft_node_is_active(node)) {
return 0;
}
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
assert(node); assert(node);
assert(node != me->node); assert(node != me->node);
if (!(me->cb.send_appendentries)) if (!raft_node_is_active(node)) {
return -1; return 0;
}
msg_appendentries_t ae = {};
ae.term = me->current_term;
ae.leader_commit = raft_get_commit_idx(me_);
ae.prev_log_idx = 0;
ae.prev_log_term = 0;
ae.msg_id = me->msg_id;
raft_index_t next_idx = raft_node_get_next_idx(node); raft_index_t next_idx = raft_node_get_next_idx(node);
/* figure out if the client needs a snapshot sent */ /* figure out if the client needs a snapshot sent */
if (0 < me->snapshot_last_idx && next_idx < me->snapshot_last_idx) if (me->snapshot_last_idx > 0 && next_idx <= me->snapshot_last_idx)
{ {
if (me->cb.send_snapshot) if (me->cb.send_snapshot)
me->cb.send_snapshot(me_, me->udata, node); me->cb.send_snapshot(me_, me->udata, node);
return RAFT_ERR_NEEDS_SNAPSHOT; return RAFT_ERR_NEEDS_SNAPSHOT;
} }
if (!me->cb.send_appendentries)
return -1;
msg_appendentries_t ae = {
.term = me->current_term,
.leader_commit = raft_get_commit_idx(me_),
.msg_id = me->msg_id,
};
ae.entries = raft_get_entries_from_idx(me_, next_idx, &ae.n_entries); ae.entries = raft_get_entries_from_idx(me_, next_idx, &ae.n_entries);
assert((!ae.entries && 0 == ae.n_entries) || assert((!ae.entries && 0 == ae.n_entries) ||
(ae.entries && 0 < ae.n_entries)); (ae.entries && 0 < ae.n_entries));
/* previous log is the log just before the new logs */ /* previous log is the log just before the new logs */
if (1 < next_idx) if (next_idx > 1)
{ {
raft_entry_t* prev_ety = raft_get_entry_from_idx(me_, next_idx - 1); raft_entry_t* prev_ety = raft_get_entry_from_idx(me_, next_idx - 1);
if (!prev_ety) if (!prev_ety)
...@@ -1498,7 +1498,7 @@ int raft_end_snapshot(raft_server_t *me_) ...@@ -1498,7 +1498,7 @@ int raft_end_snapshot(raft_server_t *me_)
/* If needed, remove compacted logs */ /* If needed, remove compacted logs */
int e = me->log_impl->poll(me->log, me->snapshot_last_idx + 1); int e = me->log_impl->poll(me->log, me->snapshot_last_idx + 1);
if (e != 0) if (e != 0)
return -1; return e;
me->snapshot_in_progress = 0; me->snapshot_in_progress = 0;
...@@ -1511,8 +1511,7 @@ int raft_end_snapshot(raft_server_t *me_) ...@@ -1511,8 +1511,7 @@ int raft_end_snapshot(raft_server_t *me_)
if (!raft_is_leader(me_)) if (!raft_is_leader(me_))
return 0; return 0;
int i; for (int i = 0; i < me->num_nodes; i++)
for (i = 0; i < me->num_nodes; i++)
{ {
raft_node_t* node = me->nodes[i]; raft_node_t* node = me->nodes[i];
...@@ -1522,7 +1521,7 @@ int raft_end_snapshot(raft_server_t *me_) ...@@ -1522,7 +1521,7 @@ int raft_end_snapshot(raft_server_t *me_)
raft_index_t next_idx = raft_node_get_next_idx(node); raft_index_t next_idx = raft_node_get_next_idx(node);
/* figure out if the client needs a snapshot sent */ /* figure out if the client needs a snapshot sent */
if (0 < me->snapshot_last_idx && next_idx < me->snapshot_last_idx) if (me->snapshot_last_idx > 0 && next_idx <= me->snapshot_last_idx)
{ {
if (me->cb.send_snapshot) if (me->cb.send_snapshot)
me->cb.send_snapshot(me_, me->udata, node); me->cb.send_snapshot(me_, me->udata, node);
......
...@@ -63,8 +63,7 @@ static int __raft_send_appendentries_capture(raft_server_t* raft, ...@@ -63,8 +63,7 @@ static int __raft_send_appendentries_capture(raft_server_t* raft,
raft_node_t* node, raft_node_t* node,
msg_appendentries_t* msg) msg_appendentries_t* msg)
{ {
msg_appendentries_t* msg_captured = (msg_appendentries_t*)udata; *((msg_appendentries_t*)udata) = *msg;
memcpy(msg_captured, msg, sizeof(msg_appendentries_t));
return 0; return 0;
} }
...@@ -486,20 +485,20 @@ void TestRaft_follower_load_from_snapshot_fails_if_log_is_newer(CuTest * tc) ...@@ -486,20 +485,20 @@ void TestRaft_follower_load_from_snapshot_fails_if_log_is_newer(CuTest * tc)
CuAssertIntEquals(tc, -1, raft_begin_load_snapshot(r, 2, 2)); CuAssertIntEquals(tc, -1, raft_begin_load_snapshot(r, 2, 2));
} }
void TestRaft_leader_sends_appendentries_when_node_next_index_was_compacted(CuTest* tc) void TestRaft_leader_sends_snapshot_when_node_next_index_was_compacted(CuTest* tc)
{ {
raft_cbs_t funcs = { raft_cbs_t funcs = {
.send_appendentries = __raft_send_appendentries_capture, .send_snapshot = __raft_send_snapshot_increment
}; };
msg_appendentries_t ae; int increment = 0;
void *r = raft_new(); void *r = raft_new();
raft_set_callbacks(r, &funcs, &ae); raft_set_callbacks(r, &funcs, &increment);
raft_node_t* node; raft_node_t* node;
raft_add_node(r, NULL, 1, 1); raft_add_node(r, NULL, 1, 1);
node = raft_add_node(r, NULL, 2, 0); raft_add_node(r, NULL, 2, 0);
raft_add_node(r, NULL, 3, 0); raft_add_node(r, NULL, 3, 0);
/* entry 1 */ /* entry 1 */
...@@ -520,6 +519,23 @@ void TestRaft_leader_sends_appendentries_when_node_next_index_was_compacted(CuTe ...@@ -520,6 +519,23 @@ void TestRaft_leader_sends_appendentries_when_node_next_index_was_compacted(CuTe
raft_set_state(r, RAFT_STATE_LEADER); raft_set_state(r, RAFT_STATE_LEADER);
raft_set_current_term(r, 2); raft_set_current_term(r, 2);
/* verify snapshot is sent */
int rc = raft_send_appendentries(r, node);
CuAssertIntEquals(tc, RAFT_ERR_NEEDS_SNAPSHOT, rc);
CuAssertIntEquals(tc, 1, increment);
/* update callbacks, verify correct appendreq is sent after the snapshot */
funcs = (raft_cbs_t) {
.send_appendentries = __raft_send_appendentries_capture,
};
msg_appendentries_t ae;
raft_set_callbacks(r, &funcs, &ae);
/* node wants an entry just one after the snapshot index */
raft_node_set_next_idx(node, raft_get_current_idx(r) + 1);
CuAssertIntEquals(tc, 0, raft_send_appendentries(r, node)); CuAssertIntEquals(tc, 0, raft_send_appendentries(r, node));
CuAssertIntEquals(tc, 2, ae.term); CuAssertIntEquals(tc, 2, ae.term);
CuAssertIntEquals(tc, 3, ae.prev_log_idx); CuAssertIntEquals(tc, 3, ae.prev_log_idx);
...@@ -652,7 +668,7 @@ void TestRaft_leader_sends_appendentries_with_correct_prev_log_idx_when_snapshot ...@@ -652,7 +668,7 @@ void TestRaft_leader_sends_appendentries_with_correct_prev_log_idx_when_snapshot
*/ */
raft_node_t* p = raft_add_node(r, NULL, 2, 0); raft_node_t* p = raft_add_node(r, NULL, 2, 0);
CuAssertTrue(tc, NULL != p); CuAssertTrue(tc, NULL != p);
raft_node_set_next_idx(p, 4); raft_node_set_next_idx(p, 5);
/* receive appendentries messages */ /* receive appendentries messages */
raft_send_appendentries(r, p); raft_send_appendentries(r, p);
...@@ -783,7 +799,7 @@ int main(void) ...@@ -783,7 +799,7 @@ int main(void)
SUITE_ADD_TEST(suite, TestRaft_follower_load_from_snapshot_fails_if_already_loaded); SUITE_ADD_TEST(suite, TestRaft_follower_load_from_snapshot_fails_if_already_loaded);
SUITE_ADD_TEST(suite, TestRaft_follower_load_from_snapshot_does_not_break_cluster_safety); SUITE_ADD_TEST(suite, TestRaft_follower_load_from_snapshot_does_not_break_cluster_safety);
SUITE_ADD_TEST(suite, TestRaft_follower_load_from_snapshot_fails_if_log_is_newer); SUITE_ADD_TEST(suite, TestRaft_follower_load_from_snapshot_fails_if_log_is_newer);
SUITE_ADD_TEST(suite, TestRaft_leader_sends_appendentries_when_node_next_index_was_compacted); SUITE_ADD_TEST(suite, TestRaft_leader_sends_snapshot_when_node_next_index_was_compacted);
SUITE_ADD_TEST(suite, TestRaft_recv_entry_fails_if_snapshot_in_progress); SUITE_ADD_TEST(suite, TestRaft_recv_entry_fails_if_snapshot_in_progress);
SUITE_ADD_TEST(suite, TestRaft_recv_entry_succeeds_if_snapshot_nonblocking_apply_is_set); SUITE_ADD_TEST(suite, TestRaft_recv_entry_succeeds_if_snapshot_nonblocking_apply_is_set);
SUITE_ADD_TEST(suite, TestRaft_follower_recv_appendentries_is_successful_when_previous_log_idx_equals_snapshot_last_idx); SUITE_ADD_TEST(suite, TestRaft_follower_recv_appendentries_is_successful_when_previous_log_idx_equals_snapshot_last_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