feat(raft-log-replication): raft_send_append_entries AppendEntries emit (tasks 4.1, 4.2)

Leader-only emit (raft.tla AppendEntries:204, Req 3.5 — no-op off leader);
prev_log_index=next_index[j]-1, prev_log_term via on_get_entry_term (0/unqueried
at index 0, Req 3.2-3.3); n_entries=1 when next_index[j]<=log_len else 0 heartbeat
(Req 3.4); leader_commit opaque passthrough. Trace fires only on successful send
(consistency with raft_log_append) and non-leader RAFT_OK asserted — both from
review. 4 spy-based tests. ASAN green.

.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_

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;
+}

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);