diff --git a/.kiro/specs/raft-log-replication/tasks.md b/.kiro/specs/raft-log-replication/tasks.md index 09ffa2eb..c495da27 100644 --- a/.kiro/specs/raft-log-replication/tasks.md +++ b/.kiro/specs/raft-log-replication/tasks.md @@ -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_ diff --git a/src/lib/raft/raft.c b/src/lib/raft/raft.c index b9fcfd19..19b7874b 100644 --- a/src/lib/raft/raft.c +++ b/src/lib/raft/raft.c @@ -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; diff --git a/src/lib/raft/raft.h b/src/lib/raft/raft.h index 8a86b70a..ca049ffe 100644 --- a/src/lib/raft/raft.h +++ b/src/lib/raft/raft.h @@ -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; /* diff --git a/src/lib/raft/raft_sim.c b/src/lib/raft/raft_sim.c index 911a3b4c..ed44de21 100644 --- a/src/lib/raft/raft_sim.c +++ b/src/lib/raft/raft_sim.c @@ -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; diff --git a/src/lib/testsuite/test_raft.c b/src/lib/testsuite/test_raft.c index 1c7b14e3..8c00fc46 100644 --- a/src/lib/testsuite/test_raft.c +++ b/src/lib/testsuite/test_raft.c @@ -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);