diff --git a/.kiro/specs/raft-log-replication/tasks.md b/.kiro/specs/raft-log-replication/tasks.md index e251091..a9df95c 100644 --- a/.kiro/specs/raft-log-replication/tasks.md +++ b/.kiro/specs/raft-log-replication/tasks.md @@ -65,14 +65,14 @@ ## 4. AppendEntries emission — raft_send_append_entries (raft.tla:204) -- [ ] 4.1 RED: emit tests (prevLogIndex/prevLogTerm, ≤1 entry, leader-only) +- [x] 4.1 RED: emit tests (prevLogIndex/prevLogTerm, ≤1 entry, leader-only) - Test: for peer `j`, emit computes `prevLogIndex = next_index[j]-1` and `prevLogTerm` via `on_get_entry_term` (0 when `prevLogIndex==0`), sends ≤1 entry, passes `leader_commit` through opaquely; a non-leader emits nothing - Observable completion: tests compile and FAIL - _Requirements: 3.2, 3.3, 3.4, 3.5_ - _Boundary: test_raft.c_ - _Depends: 2.2_ -- [ ] 4.2 GREEN: implement `raft_send_append_entries` +- [x] 4.2 GREEN: implement `raft_send_append_entries` - Compute prev index/term; `lastEntry = Min(log_len, next_index[j])`; one entry or heartbeat; call `on_send_append_entries` with the opaque `leader_commit`; leader-only guard - Observable completion: 4.1 tests pass; heartbeat (0 entries) emitted when `next_index[j] > log_len` - _Requirements: 3.2, 3.3, 3.4, 3.5_ diff --git a/src/lib/raft/raft.c b/src/lib/raft/raft.c index 336e953..bc2f5f8 100644 --- a/src/lib/raft/raft.c +++ b/src/lib/raft/raft.c @@ -503,3 +503,41 @@ int raft_log_append(raft_state_t *s, const void *entry, size_t len, return RAFT_OK; } + +int raft_send_append_entries(raft_state_t *s, int peer_idx, + uint64_t leader_commit) +{ + /* Leader-only (Req 3.5): AppendEntries is enabled only at the leader in + * raft.tla. Off the leader this is a no-op — no callback, no state change. */ + if (s->role != RAFT_LEADER) + return RAFT_OK; + + /* prevLogIndex = nextIndex[j] - 1; prevLogTerm = on_get_entry_term(prevIdx) + * when prevIdx > 0, else 0 (the "absent" sentinel, matching raft.tla + * prevLogTerm == 0 when prevLogIndex == 0). Do NOT query the callback at + * index 0 (Req 3.2). */ + uint64_t prev_log_index = s->next_index[peer_idx] - 1; + uint64_t prev_log_term = (prev_log_index > 0) + ? s->cbs->on_get_entry_term(s->ctx, prev_log_index) + : 0; + + /* lastEntry = Min(log_len, next_index[j]); send SubSeq(log, next_index[j], + * lastEntry) — at most one entry. When next_index[j] > log_len there is + * nothing to send → a 0-entry heartbeat (Req 3.3). */ + int n_entries = (s->next_index[peer_idx] <= s->log_len) ? 1 : 0; + + /* leader_commit is passed through opaquely; this spec never interprets or + * advances commit (Req 3.4 — commit is owned by raft-commit-safety). */ + int rc = s->cbs->on_send_append_entries(s->ctx, peer_idx, prev_log_index, + prev_log_term, leader_commit, + n_entries); + if (rc != 0) + return rc; + + /* Trace only on a successful send (mirrors raft_log_append, which traces + * after a confirmed persist) — a failed emit must not look successful. */ + if (s->cbs->on_trace) + s->cbs->on_trace(s->ctx, "ae_send", NULL); + + return RAFT_OK; +} diff --git a/src/lib/testsuite/test_raft.c b/src/lib/testsuite/test_raft.c index 38e6cb6..e6fa539 100644 --- a/src/lib/testsuite/test_raft.c +++ b/src/lib/testsuite/test_raft.c @@ -1690,6 +1690,207 @@ static void test_log_append_persist_failure_propagates(void) TEST_ASSERT_EQUAL_UINT64(4, s.log_len); } +/* ---------- raft_send_append_entries tests (task 4.1/4.2, raft.tla AppendEntries 204) ---------- */ + +/* Spy on_send_append_entries: records the last call's arguments and a call + * count, so emission of (peer_idx, prev_log_index, prev_log_term, leader_commit, + * n_entries) can be verified exactly. */ +typedef struct { + int call_count; + int last_peer_idx; + uint64_t last_prev_log_index; + uint64_t last_prev_log_term; + uint64_t last_leader_commit; + int last_n_entries; +} send_ae_spy_t; + +static send_ae_spy_t g_ae_spy; + +static void reset_ae_spy(void) +{ + memset(&g_ae_spy, 0, sizeof(g_ae_spy)); +} + +static int spy_send_append_entries(void *ctx, int peer_idx, + uint64_t prev_log_index, + uint64_t prev_log_term, + uint64_t leader_commit, int n_entries) +{ + (void)ctx; + g_ae_spy.call_count++; + g_ae_spy.last_peer_idx = peer_idx; + g_ae_spy.last_prev_log_index = prev_log_index; + g_ae_spy.last_prev_log_term = prev_log_term; + g_ae_spy.last_leader_commit = leader_commit; + g_ae_spy.last_n_entries = n_entries; + return 0; +} + +/* Spy on_get_entry_term: returns a configurable term per 1-based index so the + * prev_log_term computation in the emit helper can be verified. An index past + * the populated range (or index 0) returns 0 = "absent". */ +typedef struct { + int call_count; + uint64_t last_index; + uint64_t term_at[RAFT_SIM_MAX_LOG + 1]; /* 1-based: term_at[i] = term of entry i */ +} get_term_spy_t; + +static get_term_spy_t g_gt_spy; + +static void reset_gt_spy(void) +{ + memset(&g_gt_spy, 0, sizeof(g_gt_spy)); +} + +static uint64_t spy_get_entry_term(void *ctx, uint64_t index) +{ + (void)ctx; + g_gt_spy.call_count++; + g_gt_spy.last_index = index; + if (index >= 1 && index <= RAFT_SIM_MAX_LOG) + return g_gt_spy.term_at[index]; + return 0; +} + +/* + * test_send_ae_prev_index_zero_heartbeat (Req 3.2, 3.3, 3.4) + * + * next_index[j] == 1 (so prev_log_index == 0) AND next_index[j] > log_len + * (empty log, log_len == 0) → heartbeat: prev_log_index == 0, prev_log_term == 0 + * (NOT queried / treated as absent), n_entries == 0, leader_commit passed opaque. + */ +static void test_send_ae_prev_index_zero_heartbeat(void) +{ + raft_state_t s; + raft_callbacks_t cbs = make_mock_cbs(); + cbs.on_send_append_entries = spy_send_append_entries; + cbs.on_get_entry_term = spy_get_entry_term; + + reset_mock(); + reset_ae_spy(); + reset_gt_spy(); + /* empty restored log → log_len = 0, next_index[j] becomes 1 on leader */ + TEST_ASSERT_EQUAL_INT(0, init_node_log_len(&s, &cbs, 3, 0, 0)); + drive_to_leader(&s); + TEST_ASSERT_EQUAL_UINT64(0, s.log_len); + TEST_ASSERT_EQUAL_UINT64(1, s.next_index[1]); /* log_len+1 */ + + int rc = raft_send_append_entries(&s, 1, /*leader_commit=*/0); + + TEST_ASSERT_EQUAL_INT(RAFT_OK, rc); + TEST_ASSERT_EQUAL_INT(1, g_ae_spy.call_count); + TEST_ASSERT_EQUAL_INT(1, g_ae_spy.last_peer_idx); + TEST_ASSERT_EQUAL_UINT64(0, g_ae_spy.last_prev_log_index); + TEST_ASSERT_EQUAL_UINT64(0, g_ae_spy.last_prev_log_term); /* absent when prevIdx==0 */ + TEST_ASSERT_EQUAL_UINT64(0, g_ae_spy.last_leader_commit); + TEST_ASSERT_EQUAL_INT(0, g_ae_spy.last_n_entries); /* heartbeat */ + /* prev_log_index == 0 → on_get_entry_term must NOT be queried for it */ + TEST_ASSERT_EQUAL_INT(0, g_gt_spy.call_count); +} + +/* + * test_send_ae_heartbeat_when_next_past_log (Req 3.2, 3.3, 3.4) + * + * next_index[j] > log_len with prev_log_index > 0 → heartbeat (n_entries == 0). + * prev_log_index = next_index[j]-1; prev_log_term queried via on_get_entry_term. + * leader_commit passed through opaquely. + */ +static void test_send_ae_heartbeat_when_next_past_log(void) +{ + raft_state_t s; + raft_callbacks_t cbs = make_mock_cbs(); + cbs.on_send_append_entries = spy_send_append_entries; + cbs.on_get_entry_term = spy_get_entry_term; + + reset_mock(); + reset_ae_spy(); + reset_gt_spy(); + /* restored log_len = 4; on becoming leader next_index[j] = 5 (= log_len+1), + * so next_index[j] > log_len → heartbeat, prev_log_index = 4 (> 0). */ + TEST_ASSERT_EQUAL_INT(0, init_node_log_len(&s, &cbs, 3, 0, 4)); + drive_to_leader(&s); + uint64_t leader_term = s.term; + g_gt_spy.term_at[4] = leader_term; /* term of entry at prev_log_index = 4 */ + + int rc = raft_send_append_entries(&s, 1, /*leader_commit=*/42); + + TEST_ASSERT_EQUAL_INT(RAFT_OK, rc); + TEST_ASSERT_EQUAL_INT(1, g_ae_spy.call_count); + TEST_ASSERT_EQUAL_INT(1, g_ae_spy.last_peer_idx); + TEST_ASSERT_EQUAL_UINT64(4, g_ae_spy.last_prev_log_index); /* next_index-1 */ + TEST_ASSERT_EQUAL_UINT64(leader_term, g_ae_spy.last_prev_log_term); + TEST_ASSERT_EQUAL_UINT64(42, g_ae_spy.last_leader_commit); + TEST_ASSERT_EQUAL_INT(0, g_ae_spy.last_n_entries); /* heartbeat */ + /* prev_log_index = 4 (> 0) → on_get_entry_term queried for index 4 */ + TEST_ASSERT_EQUAL_UINT64(4, g_gt_spy.last_index); + TEST_ASSERT_TRUE(g_gt_spy.call_count >= 1); +} + +/* + * test_send_ae_one_entry_when_next_within_log (Req 3.2, 3.3, 3.4) + * + * next_index[j] <= log_len → one entry (n_entries == 1). + * prev_log_index = next_index[j]-1; prev_log_term queried via on_get_entry_term. + */ +static void test_send_ae_one_entry_when_next_within_log(void) +{ + raft_state_t s; + raft_callbacks_t cbs = make_mock_cbs(); + cbs.on_send_append_entries = spy_send_append_entries; + cbs.on_get_entry_term = spy_get_entry_term; + + reset_mock(); + reset_ae_spy(); + reset_gt_spy(); + /* restored log_len = 5; on becoming leader next_index[j] = 6. Walk one peer + * back to next_index = 3 (<= log_len) so one entry (index 3) must be sent; + * prev_log_index = 2 (> 0). */ + TEST_ASSERT_EQUAL_INT(0, init_node_log_len(&s, &cbs, 3, 0, 5)); + drive_to_leader(&s); + uint64_t leader_term = s.term; + s.next_index[1] = 3; /* construct the within-log case */ + g_gt_spy.term_at[2] = leader_term; /* term of entry at prev_log_index = 2 */ + + int rc = raft_send_append_entries(&s, 1, /*leader_commit=*/7); + + TEST_ASSERT_EQUAL_INT(RAFT_OK, rc); + TEST_ASSERT_EQUAL_INT(1, g_ae_spy.call_count); + TEST_ASSERT_EQUAL_INT(1, g_ae_spy.last_peer_idx); + TEST_ASSERT_EQUAL_UINT64(2, g_ae_spy.last_prev_log_index); /* next_index-1 */ + TEST_ASSERT_EQUAL_UINT64(leader_term, g_ae_spy.last_prev_log_term); + TEST_ASSERT_EQUAL_UINT64(7, g_ae_spy.last_leader_commit); + TEST_ASSERT_EQUAL_INT(1, g_ae_spy.last_n_entries); /* one entry */ + TEST_ASSERT_EQUAL_UINT64(2, g_gt_spy.last_index); +} + +/* + * test_send_ae_non_leader_emits_nothing (Req 3.5) + * + * A non-leader (FOLLOWER) raft_send_append_entries must NOT call + * on_send_append_entries at all. + */ +static void test_send_ae_non_leader_emits_nothing(void) +{ + raft_state_t s; + raft_callbacks_t cbs = make_mock_cbs(); + cbs.on_send_append_entries = spy_send_append_entries; + cbs.on_get_entry_term = spy_get_entry_term; + + reset_mock(); + reset_ae_spy(); + reset_gt_spy(); + TEST_ASSERT_EQUAL_INT(0, init_node_log_len(&s, &cbs, 3, 0, 4)); + /* raft_init leaves the node RAFT_FOLLOWER */ + TEST_ASSERT_EQUAL_INT(RAFT_FOLLOWER, s.role); + + int rc = raft_send_append_entries(&s, 1, /*leader_commit=*/9); + + /* No emission whatsoever; the helper is a no-op off the leader. */ + TEST_ASSERT_EQUAL_INT(0, g_ae_spy.call_count); + /* Header contract: returns RAFT_OK on every path, including non-leader. */ + TEST_ASSERT_EQUAL_INT(RAFT_OK, rc); +} + /* ---------- raft_on_election_timeout tests (task 3.5) ---------- */ /* @@ -4294,6 +4495,12 @@ void test_raft(void) RUN_TEST(test_log_append_candidate_rejected); RUN_TEST(test_log_append_persist_failure_propagates); + /* raft_send_append_entries (task 4.1/4.2) */ + RUN_TEST(test_send_ae_prev_index_zero_heartbeat); + RUN_TEST(test_send_ae_heartbeat_when_next_past_log); + RUN_TEST(test_send_ae_one_entry_when_next_within_log); + RUN_TEST(test_send_ae_non_leader_emits_nothing); + /* raft_on_vote_request (task 3.3) */ RUN_TEST(test_vote_req_stale_term); RUN_TEST(test_vote_req_higher_term_updates_state);