feat(raft-log-replication): require 4 log callbacks in raft_init + sim log backing (tasks 2.1, 8.1)

2.1: raft_init returns -EINVAL when any of on_persist_entry/on_get_entry_term/
on_send_append_entries/on_truncate_from is NULL (Req 7.1-7.4), before any write
to *s; restores s->log_len = cfg->initial_log_len (Req 1.1).

8.1: raft_sim gains a per-node in-memory log (term+bytes per 1-based index) and
an AE buffer queue; wires all four callbacks (persist/get-term/truncate/send).
Pulled forward to land with 2.1 so the stricter init does not leave the
sim/proto suite red between tasks. AE delivery deferred to 8.2.

Suite green under ASAN (Ok:1 Fail:0).
This commit is contained in:
Rongsong Shen 2026-06-21 15:30:09 +00:00
parent 7b6cbb8e3e
commit d70f2507c3
5 changed files with 314 additions and 2 deletions

View file

@ -31,7 +31,7 @@
## 2. Init validation and leader-progress initialization
- [ ] 2.1 RED+GREEN: `raft_init` requires the four new callbacks
- [x] 2.1 RED+GREEN: `raft_init` requires the four new callbacks
- Test: `raft_init` returns `-EINVAL` when any of the four new callbacks is NULL; returns 0 when all present
- Implement the NULL checks in `raft_init` (mirroring the existing required-callback checks); restore `s->log_len = cfg->initial_log_len`
- Observable completion: new `-EINVAL` test cases pass; existing election init tests still green
@ -136,7 +136,8 @@
## 8. Simulator backing: in-memory log + AppendEntries routing
- [ ] 8.1 Add an in-memory per-node log and wire the four callbacks in `raft_sim.c`
- [x] 8.1 Add an in-memory per-node log and wire the four callbacks in `raft_sim.c`
- NOTE (impl 2026-06-21): pulled forward and landed together with 2.1 — task 2.1's stricter `raft_init` (four callbacks now REQUIRED) makes `raft_sim_init` return -EINVAL, so ~15 `test_sim_*`/`test_proto_*` tests go red the instant 2.1 lands. 8.1's callback wiring only needs the header signatures (not 6.3/7.2), so it compiles standalone; co-committing 2.1+8.1 keeps the suite green at every commit. AE delivery (8.2) still deferred.
- Add a per-`raft_sim_node_t` in-memory entry array (`{term, bytes}` per index); implement `on_persist_entry` (append/overwrite at index), `on_get_entry_term` (term or 0), `on_truncate_from` (drop index..end), and buffer `on_send_append_entries` into a new AE RPC queue
- Observable completion: a sim node can append, read a term back, and truncate deterministically; `raft_sim_init` wires the four callbacks for every node
- _Requirements: 8.2, 7.5_

View file

@ -36,6 +36,17 @@ int raft_init(raft_state_t *s, const raft_config_t *cfg)
return -EINVAL;
if (!cfg->cbs->on_start_election_after_delay)
return -EINVAL;
/* Log-replication required callbacks (raft-log-replication Req 7.1–7.4):
* all four must be non-NULL — the library owns the protocol decisions and
* delegates the I/O (persist/read-term/send/truncate) through them. */
if (!cfg->cbs->on_persist_entry)
return -EINVAL;
if (!cfg->cbs->on_get_entry_term)
return -EINVAL;
if (!cfg->cbs->on_send_append_entries)
return -EINVAL;
if (!cfg->cbs->on_truncate_from)
return -EINVAL;
s->term = cfg->initial_term;
s->role = RAFT_FOLLOWER;
@ -52,6 +63,11 @@ int raft_init(raft_state_t *s, const raft_config_t *cfg)
s->election_decided = 0;
s->election_timeout_ms = cfg->election_timeout_ms ? cfg->election_timeout_ms : 2000;
s->my_priority = cfg->my_priority;
/* Log length restored from the adapter's recovered WAL length; the library
* never reads storage (same discipline as initial_term/voted_for). 0 on
* first boot. (raft-log-replication Req 1.1; design "log_len restored at
* init from initial_log_len".) */
s->log_len = cfg->initial_log_len;
s->cbs = cfg->cbs;
s->ctx = cfg->ctx;

View file

@ -370,6 +370,15 @@ void raft_snapshot(const raft_state_t *s, raft_snapshot_t *snap);
#define RAFT_SIM_MAX_NODES 8
/* Upper bound on a simulated node's in-memory log. Test entries are tiny and
* the sim drives short runs, so a small fixed capacity keeps the node struct
* ASAN-clean (no allocation) while comfortably exceeding any test's log depth. */
#define RAFT_SIM_MAX_LOG 64
/* Fixed entry payload size — test entries are a few bytes; bounded copies only. */
#define RAFT_SIM_ENTRY_LEN 16
/* Upper bound on buffered AppendEntries awaiting delivery (mirrors pending[]). */
#define RAFT_SIM_MAX_AE 64
/* Buffered RPC entry: one pending vote-request to route */
typedef struct {
int src;
@ -377,6 +386,32 @@ typedef struct {
uint64_t term;
} raft_sim_rpc_t;
/* One in-memory log entry: term stamp + fixed-size payload bytes.
* term == 0 means "absent" (matches the 0-means-absent convention of
* on_get_entry_term); 1-based indexing maps slot[i-1] to log index i. */
typedef struct {
uint64_t term;
unsigned char bytes[RAFT_SIM_ENTRY_LEN];
size_t len;
} raft_sim_entry_t;
/* Buffered AppendEntries RPC awaiting synchronous delivery (task 8.2).
* For n_entries == 1 the single entry's index/term/bytes are captured here
* because the library holds no entry bytes — the sender sources them from its
* own in-memory log at prev_log_index + 1 when buffering. */
typedef struct {
int src;
int dst;
uint64_t prev_log_index;
uint64_t prev_log_term;
uint64_t leader_commit;
int n_entries; /* 0 (heartbeat) or 1 (this spec) */
uint64_t entry_index; /* valid when n_entries == 1 */
uint64_t entry_term; /* valid when n_entries == 1 */
unsigned char entry_bytes[RAFT_SIM_ENTRY_LEN];
size_t entry_len; /* valid when n_entries == 1 */
} raft_sim_ae_t;
/* One simulated node */
typedef struct {
raft_state_t state;
@ -384,6 +419,10 @@ typedef struct {
void *ctx; /* points back to raft_sim_cluster_t */
int idx;
int partition; /* 1 = isolated from other nodes */
/* in-memory replicated log backing the four log callbacks (1-based index;
* log[i-1] holds entry i; log_count is the highest occupied index). */
raft_sim_entry_t log[RAFT_SIM_MAX_LOG];
int log_count;
/* per-node trace event log for test assertions */
char trace_buf[4096];
int trace_len;
@ -397,6 +436,10 @@ typedef struct {
/* buffered RPC queue */
raft_sim_rpc_t pending[64];
int pending_count;
/* buffered AppendEntries queue (delivered by raft_sim_deliver_append_entries,
* task 8.2); on_send_append_entries enqueues here. */
raft_sim_ae_t ae_pending[RAFT_SIM_MAX_AE];
int ae_pending_count;
} raft_sim_cluster_t;
/*

View file

@ -69,6 +69,127 @@ static void sim_become_leader(void *ctx, uint64_t term)
c->leader_count++;
}
/* ---------- log-replication callbacks (raft-log-replication) ----------
*
* These back the four required log callbacks with the node's in-memory log.
* ctx is raft_sim_node_t * (same convention as the wired election callbacks).
* The library holds no entry bytes, so the simulator owns storage here:
* - on_persist_entry: append/overwrite at 1-based index (term + bytes)
* - on_get_entry_term: read the stored term (0 = absent)
* - on_truncate_from: drop entries index..end
* - on_send_append_entries: buffer one AE (sourcing the single entry from the
* sender's own log at prev_log_index+1 when n_entries == 1)
* No AppendEntries delivery happens here — that is raft_sim_deliver_append_entries
* (task 8.2). These callbacks only do storage/term-read/truncate/buffer.
*/
/*
* sim_persist_entry: durably append or overwrite the entry at 1-based `index`.
* Sets the slot's term and copies up to RAFT_SIM_ENTRY_LEN bytes; extends
* log_count when appending past the current tail.
*/
static int sim_persist_entry(void *ctx, uint64_t index, uint64_t term,
const void *entry, size_t len)
{
raft_sim_node_t *node = (raft_sim_node_t *)ctx;
if (index < 1 || index > RAFT_SIM_MAX_LOG)
return -1; /* out of fixed sim bounds */
raft_sim_entry_t *slot = &node->log[index - 1];
slot->term = term;
size_t n = len;
if (n > RAFT_SIM_ENTRY_LEN)
n = RAFT_SIM_ENTRY_LEN; /* bounded copy */
if (entry && n > 0)
memcpy(slot->bytes, entry, n);
slot->len = n;
if ((int)index > node->log_count)
node->log_count = (int)index;
return 0;
}
/*
* sim_get_entry_term: return the stored term at 1-based `index`, or 0 if absent
* (0-means-absent convention).
*/
static uint64_t sim_get_entry_term(void *ctx, uint64_t index)
{
raft_sim_node_t *node = (raft_sim_node_t *)ctx;
if (index < 1 || index > (uint64_t)node->log_count)
return 0;
return node->log[index - 1].term;
}
/*
* sim_truncate_from: drop entries at `index`..end (clear term so they read as
* absent) and shrink log_count.
*/
static int sim_truncate_from(void *ctx, uint64_t index)
{
raft_sim_node_t *node = (raft_sim_node_t *)ctx;
if (index < 1)
return -1;
for (uint64_t i = index; i <= (uint64_t)node->log_count; i++) {
if (i <= RAFT_SIM_MAX_LOG) {
raft_sim_entry_t *slot = &node->log[i - 1];
memset(slot, 0, sizeof(*slot)); /* term=0 => absent */
}
}
if (node->log_count >= (int)index)
node->log_count = (int)index - 1;
return 0;
}
/*
* sim_send_append_entries: buffer one AppendEntries into the cluster AE queue.
* For n_entries == 1 the single entry's index/term/bytes are sourced from the
* sender's own in-memory log at prev_log_index + 1 (the library holds no bytes).
*/
static int sim_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)
{
raft_sim_node_t *node = (raft_sim_node_t *)ctx;
raft_sim_cluster_t *c = (raft_sim_cluster_t *)node->ctx;
if (c->ae_pending_count >= RAFT_SIM_MAX_AE)
return -1; /* queue full — drop */
raft_sim_ae_t *ae = &c->ae_pending[c->ae_pending_count++];
memset(ae, 0, sizeof(*ae));
ae->src = node->idx;
ae->dst = peer_idx;
ae->prev_log_index = prev_log_index;
ae->prev_log_term = prev_log_term;
ae->leader_commit = leader_commit;
ae->n_entries = n_entries;
if (n_entries == 1) {
uint64_t idx = prev_log_index + 1;
ae->entry_index = idx;
if (idx >= 1 && idx <= (uint64_t)node->log_count) {
raft_sim_entry_t *slot = &node->log[idx - 1];
ae->entry_term = slot->term;
ae->entry_len = slot->len;
if (slot->len > 0)
memcpy(ae->entry_bytes, slot->bytes, slot->len);
}
}
return 0;
}
/*
* sim_on_trace: append the event string to the node's trace_buf.
*
@ -125,6 +246,11 @@ int raft_sim_init(raft_sim_cluster_t *c, int n_nodes, uint32_t election_timeout_
node->cbs.on_schedule_timeout = sim_noop_schedule_timeout;
node->cbs.on_cancel_timeout = sim_noop_cancel_timeout;
node->cbs.on_start_election_after_delay = sim_noop_start_election_after_delay;
/* log-replication callbacks backed by the node's in-memory log */
node->cbs.on_persist_entry = sim_persist_entry;
node->cbs.on_get_entry_term = sim_get_entry_term;
node->cbs.on_send_append_entries = sim_send_append_entries;
node->cbs.on_truncate_from = sim_truncate_from;
node->cbs.on_vote_eligible = NULL; /* optional — treat all as eligible */
node->cbs.on_trace = sim_on_trace;

View file

@ -46,6 +46,37 @@ static void stub_start_election_after_delay(void *ctx, uint32_t delay_ms)
(void)ctx; (void)delay_ms;
}
/* ---- log-replication callback stubs (raft-log-replication) ---- */
static int stub_persist_entry(void *ctx, uint64_t index, uint64_t term,
const void *entry, size_t len)
{
(void)ctx; (void)index; (void)term; (void)entry; (void)len;
return 0;
}
static uint64_t stub_get_entry_term(void *ctx, uint64_t index)
{
(void)ctx; (void)index;
return 0;
}
static int stub_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; (void)peer_idx; (void)prev_log_index; (void)prev_log_term;
(void)leader_commit; (void)n_entries;
return 0;
}
static int stub_truncate_from(void *ctx, uint64_t index)
{
(void)ctx; (void)index;
return 0;
}
/* Build a fully-populated callbacks struct */
static raft_callbacks_t make_full_cbs(void)
{
@ -58,6 +89,10 @@ static raft_callbacks_t make_full_cbs(void)
cbs.on_schedule_timeout = stub_schedule_timeout;
cbs.on_cancel_timeout = stub_cancel_timeout;
cbs.on_start_election_after_delay = stub_start_election_after_delay;
cbs.on_persist_entry = stub_persist_entry;
cbs.on_get_entry_term = stub_get_entry_term;
cbs.on_send_append_entries = stub_send_append_entries;
cbs.on_truncate_from = stub_truncate_from;
/* on_vote_eligible and on_trace left NULL (optional) */
return cbs;
}
@ -222,6 +257,12 @@ static raft_callbacks_t make_mock_cbs(void)
cbs.on_cancel_timeout = mock_cancel_timeout;
cbs.on_start_election_after_delay = mock_start_election_after_delay;
cbs.on_send_pre_vote_request = mock_send_pre_vote_request;
/* Log-replication callbacks: stubs (task 2.1 only needs them non-NULL for
* raft_init validation; behaviour is exercised by later tasks). */
cbs.on_persist_entry = stub_persist_entry;
cbs.on_get_entry_term = stub_get_entry_term;
cbs.on_send_append_entries = stub_send_append_entries;
cbs.on_truncate_from = stub_truncate_from;
/* on_vote_eligible and on_trace left NULL — tests that need them set explicitly */
return cbs;
}
@ -361,6 +402,86 @@ static void test_raft_init_null_start_election_after_delay_returns_einval(void)
TEST_ASSERT_EQUAL_INT(-EINVAL, raft_init(&s, &cfg));
}
/* ---- log-replication required callbacks (Req 7.1–7.4): each NULL → -EINVAL ---- */
static void test_raft_init_null_persist_entry_returns_einval(void)
{
raft_state_t s;
raft_callbacks_t cbs = make_full_cbs();
raft_config_t cfg;
memset(&s, 0, sizeof(s));
cbs.on_persist_entry = NULL;
memset(&cfg, 0, sizeof(cfg));
cfg.n_peers = 3; cfg.self_idx = 0; cfg.cbs = &cbs;
TEST_ASSERT_EQUAL_INT(-EINVAL, raft_init(&s, &cfg));
}
static void test_raft_init_null_get_entry_term_returns_einval(void)
{
raft_state_t s;
raft_callbacks_t cbs = make_full_cbs();
raft_config_t cfg;
memset(&s, 0, sizeof(s));
cbs.on_get_entry_term = NULL;
memset(&cfg, 0, sizeof(cfg));
cfg.n_peers = 3; cfg.self_idx = 0; cfg.cbs = &cbs;
TEST_ASSERT_EQUAL_INT(-EINVAL, raft_init(&s, &cfg));
}
static void test_raft_init_null_send_append_entries_returns_einval(void)
{
raft_state_t s;
raft_callbacks_t cbs = make_full_cbs();
raft_config_t cfg;
memset(&s, 0, sizeof(s));
cbs.on_send_append_entries = NULL;
memset(&cfg, 0, sizeof(cfg));
cfg.n_peers = 3; cfg.self_idx = 0; cfg.cbs = &cbs;
TEST_ASSERT_EQUAL_INT(-EINVAL, raft_init(&s, &cfg));
}
static void test_raft_init_null_truncate_from_returns_einval(void)
{
raft_state_t s;
raft_callbacks_t cbs = make_full_cbs();
raft_config_t cfg;
memset(&s, 0, sizeof(s));
cbs.on_truncate_from = NULL;
memset(&cfg, 0, sizeof(cfg));
cfg.n_peers = 3; cfg.self_idx = 0; cfg.cbs = &cbs;
TEST_ASSERT_EQUAL_INT(-EINVAL, raft_init(&s, &cfg));
}
/* All four log callbacks present → succeeds; initial_log_len restored to log_len (Req 1.1) */
static void test_raft_init_log_callbacks_present_succeeds(void)
{
raft_state_t s;
raft_callbacks_t cbs = make_full_cbs();
raft_config_t cfg;
memset(&s, 0, sizeof(s));
memset(&cfg, 0, sizeof(cfg));
cfg.n_peers = 3;
cfg.self_idx = 0;
cfg.initial_log_len = 5; /* WAL length restored from disk */
cfg.cbs = &cbs;
TEST_ASSERT_EQUAL_INT(0, raft_init(&s, &cfg));
TEST_ASSERT_EQUAL_UINT64(5, s.log_len);
}
/* n_peers even (>1) → -EINVAL: even-sized clusters have no fault-tolerance
* advantage over the next-smaller odd cluster; reject to prevent misconfiguration. */
static void test_raft_init_even_peers_returns_einval(void)
@ -3864,6 +3985,11 @@ void test_raft(void)
RUN_TEST(test_raft_init_null_schedule_timeout_returns_einval);
RUN_TEST(test_raft_init_null_cancel_timeout_returns_einval);
RUN_TEST(test_raft_init_null_start_election_after_delay_returns_einval);
RUN_TEST(test_raft_init_null_persist_entry_returns_einval);
RUN_TEST(test_raft_init_null_get_entry_term_returns_einval);
RUN_TEST(test_raft_init_null_send_append_entries_returns_einval);
RUN_TEST(test_raft_init_null_truncate_from_returns_einval);
RUN_TEST(test_raft_init_log_callbacks_present_succeeds);
RUN_TEST(test_raft_init_even_peers_returns_einval);
RUN_TEST(test_raft_init_zero_peers_returns_einval);
RUN_TEST(test_raft_init_success_fields);