diff --git a/.kiro/specs/raft-adapter-conformance/tasks.md b/.kiro/specs/raft-adapter-conformance/tasks.md index 054978cf..27d89d24 100644 --- a/.kiro/specs/raft-adapter-conformance/tasks.md +++ b/.kiro/specs/raft-adapter-conformance/tasks.md @@ -16,7 +16,7 @@ - _Requirements: 2.2, 2.3, 2.5, 7.4_ - _Boundary: config_repl.c reply construction_ -- [ ] 2.2 (P) Wire term-adoption into the primary-side configuration-push reply handler +- [x] 2.2 (P) Wire term-adoption into the primary-side configuration-push reply handler - In the existing `ReplConfigPushReply` branch of `myreq_handler` (`cluster.c`, the code that already checks `cack->accepted` and calls `repl_on_config_ack` or warns-and-ignores), add a term comparison: when `cack->term` exceeds the primary's current term, call `raft_on_higher_term_observed` — the same adoption entry point the heartbeat-reply and WAL-NACK vectors already use - Do not alter the existing accept/reject branching logic itself, only add the term check alongside it - Observable: a config-push reply carrying a higher term causes the primary to adopt that term and step down from leadership held at the lower term @@ -116,3 +116,6 @@ - Task 1: adding a `$C.mapname`-tagged capnp field requires a companion edit to the hand-written C struct in `src/include/typemap.h` (the codegen writes into that struct by name; it does not synthesize it). `ReplConfigPushReply.term` required `struct repl_config_push_reply_s.term` in typemap.h — design.md's File Structure Plan didn't call this out, but it's the established, required pattern for every existing `$C.mapname` struct in that file, not scope creep. Any future task adding a capnp field should expect the same pairing. - Authoritative test verification recipe (confirmed working): `rm -rf /tmp/xce_*` then run `ASAN_OPTIONS=detect_leaks=0 src/xce.build/controller/controller_unit_test` directly and read its own stdout `N Tests M Failures` line — meson's aggregate Ok/Fail can show a spurious TIMEOUT/inline FAIL under its own harness even when the direct binary run is clean (1053 Tests 0 Failures confirmed for task 1's baseline). +- Task 2.2 (bug-1482): `myreq_handler`'s first unconditional step is `xce_decode(pkt, &request)`, which itself unconditionally `free_packet(pkt)` + frees the `io_packet_t` struct before returning on every path — `pkt` ownership transfers into `xce_decode`, the caller must never touch/free it afterward. Any new `_for_test` wrapper around `myreq_handler` (or `xce_decode` callers generally) must document this, not the reverse ("caller owns/frees pkt") — that wrong assumption is what caused the original test-only wrapper's doc comment to mislead a reviewer into requesting a double-free. +- Task 2.2: driving a test through `cluster.c`'s real `myreq_handler` (rather than testing extracted logic in isolation) requires adding `cluster.c` to `controller_unit_test_src` in `meson.build` plus link-satisfying fakes for symbols its OTHER unrelated switch-cases reference (task.c/opget.c/sched.c/apiserver.c/myconf.c) — this is pre-existing, established convention in this codebase (`test_worker_join.c`, `test_commit_barrier_reply.c` already do it for the same reason), not scope creep. Any new test driving `myreq_handler` end-to-end should expect this same footprint. +- New test-only wrappers exposing a `static` production dispatch function (the `_for_test` pattern, e.g. `peer_ctrl_msg_handler_for_test`, `myreq_handler_for_test`) MUST be wrapped in `#ifdef XCE_TESTING`/`#endif` — the source file compiles into the production binary too, and an unguarded wrapper ships as a new linked symbol in production with zero production purpose. diff --git a/src/controller/cluster.c b/src/controller/cluster.c index 239f3274..11466206 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -749,6 +749,26 @@ static int myreq_handler(vtask_t* task) && workreply->data.replconfigpushreplymsg) { repl_config_push_reply_t *cack = workreply->data.replconfigpushreplymsg; + + /* raft-adapter-conformance Req 7.4/7.5: a config-push reply now + * carries the replying member's current term (task 1 schema + * addition + task 2.1 send side) -- route it through the same + * term-discovery entry point the heartbeat-reply and WAL-NACK + * vectors already use (bug-1421; ha.c raft_lock-guarded + * raft_on_higher_term_observed calls) so a stale-term primary + * adopts the higher term and steps down instead of retrying + * forever against a standby that has already moved on. Applies + * regardless of accept/reject below -- a reject can still carry + * a useful higher term. raft_on_higher_term_observed is itself + * strictly-greater-than-only (ha-async-io.md Heartbeat Model), + * so no separate `>` comparison is needed here; raft_lock + * discipline mirrors the vote-request/pre-vote-request branches + * above (bug-1435 -- serialize against concurrent raft_state_t + * access). */ + pthread_mutex_lock(&c->ha.raft_lock); + (void)raft_on_higher_term_observed(&c->ha.raft, cack->term); + pthread_mutex_unlock(&c->ha.raft_lock); + if (cack->accepted) { repl_on_config_ack(&c->persist.repl, ep, cack->version_tag); TRACE("myreq_handler: config ACK from inbound standby" @@ -779,6 +799,39 @@ static int myreq_handler(vtask_t* task) return rc; } +#ifdef XCE_TESTING +/* + * myreq_handler_for_test: expose myreq_handler for tests (peer_ctrl_msg_ + * handler_for_test precedent, ha.c). Builds the vtask_t/vtask_data_t wrapper + * vtask_entry constructs internally in production so a test can decode+ + * dispatch a real packet through the actual switch -- including the + * ReplConfigPushReply branch's term-adoption check (raft-adapter- + * conformance Req 7.4/7.5). Packet ownership follows myreq_handler's own + * contract: myreq_handler unconditionally calls xce_decode(pkt, ...) as its + * first step, and xce_decode() itself consumes pkt -- it calls + * free_packet(pkt) (frees pkt->data) and FREEUP(pkt) (frees the io_packet_t + * struct) before returning (request.c). Ownership of `pkt` is therefore + * TRANSFERRED into this call: the caller must NOT touch or free `pkt` after + * calling myreq_handler_for_test -- doing so is a use-after-free/double-free. + * (xce_free_request separately releases the decoded xce_request_t, an + * unrelated stack-local object in myreq_handler.) + */ +int myreq_handler_for_test(endpoint_t* ep, io_packet_t* pkt) +{ + vtask_data_t data; + memset(&data, 0, sizeof(data)); + data.vtype = VTASK_IO; + data.u.iodata.ep = ep; + data.u.iodata.pkt = pkt; + + vtask_t task; + memset(&task, 0, sizeof(task)); + task.data = &data; + + return myreq_handler(&task); +} +#endif /* XCE_TESTING */ + static int vtask_entry(vtask_t* task) { int rc; diff --git a/src/controller/meson.build b/src/controller/meson.build index f7983fc9..6e00c6cf 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -193,6 +193,7 @@ controller_unit_test_src = [ 'ctrl_metrics.c', 'config_repl.c', 'slowlane.c', + 'cluster.c', ] ssl_dep = dependency('openssl') diff --git a/src/controller/testsuite/test_config_repl.c b/src/controller/testsuite/test_config_repl.c index f9248e66..f189cce3 100644 --- a/src/controller/testsuite/test_config_repl.c +++ b/src/controller/testsuite/test_config_repl.c @@ -21,6 +21,7 @@ #include +#include "cluster.h" #include "commit.h" #include "config_repl.h" #include "controller.capnp.h" @@ -3559,6 +3560,174 @@ static void test_config_push_sha256_mismatch_sends_explicit_reject_reply(void) printf("test_config_push_sha256_mismatch_sends_explicit_reject_reply: PASS\n"); } +/* raft-adapter-conformance task 2.2 (Req 7.4, 7.5): a config-push reply + * arriving on the PRIMARY's inbound standby endpoint now carries the + * replying member's current term (task 1 schema addition + task 2.1 send + * side). myreq_handler's ReplConfigPushReply branch (cluster.c) must route + * that term through raft_on_higher_term_observed -- the SAME term-discovery + * entry point the heartbeat-reply and WAL-NACK vectors already use + * (bug-1421; ha.c ~6650/~6753) -- so a stale-term primary adopts the higher + * term and steps down instead of retrying forever against a standby that + * has already moved on. + * + * Drives the REAL decode+dispatch path: builds the wire bytes exactly as + * the standby's send_config_push_reply puts them on the wire + * (config_repl.c), then pushes them through myreq_handler_for_test -- the + * production myreq_handler exposed for tests (cluster.c), same precedent as + * peer_ctrl_msg_handler_for_test in ha.c. + * + * Fixture mirrors test_intake_higher_term_heartbeat_adopts_term + * (test_term_safety_adapter.c): ha->role is left at CTRL_STANDBY (which + * ha_dispatch_gate admits OPCode_reply for unconditionally, ha.c ~4637) so + * the fixture doesn't need real quorum/replicator plumbing, while the RAFT + * LIBRARY's own role is forced to RAFT_LEADER -- so the "adopt term and + * step down from leadership" observable is scoped precisely to + * raft_on_higher_term_observed's own was_leader/step-down behavior, without + * dragging in the separate CTRL_PRIMARY-demotion side-effect chain (commit + * engine / hooks), which is a different, already-covered code path. */ +extern int myreq_handler_for_test(endpoint_t *ep, io_packet_t *pkt); + +/* Local fakes satisfying cluster.c's OTHER, unrelated OPCode-switch-case + * dependencies now that cluster.c is linked into controller_unit_test + * (meson.build -- required so myreq_handler_for_test above is reachable at + * all). None of these are exercised by the OPCode_reply/ReplConfigPushReply + * path this test drives; they exist purely to satisfy the linker for + * cluster.c's other switch cases (job/task/scheduler/apiserver/get/ + * clusterconf handling), which pull in task.c/opget.c/sched.c/apiserver.c/ + * myconf.c -- files intentionally NOT part of this binary (see the sched.c/ + * dataex.c comment above sched_push_unit_test_src, meson.build). Same + * precedent as test_worker_join.c / test_commit_barrier_reply.c's own local + * fakes for on_scheduler_join()/notify()/collect_confvars(). */ +int apiserver_init(cluster_t *c) +{ (void)c; return 0; } +int on_task_start(cluster_t *c, xce_task_pidinfo_t *info) +{ (void)c; (void)info; return 0; } +int on_task_fail(cluster_t *c, xce_task_failinfo_t *info) +{ (void)c; (void)info; return 0; } +int on_task_finish(cluster_t *c, xce_task_finish_t *info) +{ (void)c; (void)info; return 0; } +int do_get_op(cluster_t *c, endpoint_t *ep, xce_request_t *req) +{ (void)c; (void)ep; (void)req; return 0; } +int handle_scheduler_leave(cluster_t *c) +{ (void)c; return 0; } +int add_config(void *c, char *name, char *value) +{ (void)c; (void)name; (void)value; return 0; } +cluster_config_t *get_default_conf(void *cl) +{ (void)cl; return NULL; } +int handle_clusterconf(cluster_t *c, endpoint_t *ep, xce_request_t *req) +{ (void)c; (void)ep; (void)req; return 0; } +int do_runtasks(cluster_t *c, endpoint_t *ep, xce_request_t *req) +{ (void)c; (void)ep; (void)req; return 0; } +int handle_apiserver_leave(cluster_t *c, endpoint_t *ep) +{ (void)c; (void)ep; return 0; } + +static io_packet_t *cpr22_build_config_push_reply_pkt(int accepted, + uint64_t version_tag, + uint64_t term) +{ + struct capn cs; + struct capn_segment *seg; + struct ReplConfigPushReply ack_capn; + ReplConfigPushReply_ptr ack_ptr; + struct RequestReply rr_capn; + RequestReply_ptr rr_ptr; + struct Request rq_capn; + Request_ptr rq_ptr; + io_packet_t *pkt; + + capn_init_malloc(&cs); + seg = capn_root(&cs).seg; + + memset(&ack_capn, 0, sizeof(ack_capn)); + ack_capn.accepted = accepted ? 1 : 0; + ack_capn.versionTag = version_tag; + ack_capn.term = term; + ack_ptr = new_ReplConfigPushReply(seg); + write_ReplConfigPushReply(&ack_capn, ack_ptr); + + memset(&rr_capn, 0, sizeof(rr_capn)); + rr_capn.seq = 1; + rr_capn.opcode = OPCode_replConfigPush; + rr_capn.status = 0; + rr_capn.data_which = RequestReply_data_replconfigpushreplymsg; + rr_capn.data.replconfigpushreplymsg = ack_ptr; + rr_ptr = new_RequestReply(seg); + write_RequestReply(&rr_capn, rr_ptr); + + memset(&rq_capn, 0, sizeof(rq_capn)); + rq_capn.opcode = OPCode_reply; + rq_capn.seq = 1; + rq_capn.body_which = Request_body_replymsg; + rq_capn.body.replymsg = rr_ptr; + rq_ptr = new_Request(seg); + write_Request(&rq_capn, rq_ptr); + capn_setp(capn_root(&cs), 0, rq_ptr.p); + + pkt = calloc(1, sizeof(*pkt)); + TEST_ASSERT_NOT_NULL_MESSAGE(pkt, "config-push reply pkt alloc"); + pkt->sz = capn_size(&cs); + TEST_ASSERT_TRUE_MESSAGE(alloc_packet(pkt) >= 0, + "config-push reply pkt data alloc"); + capn_write_mem(&cs, (uint8_t *)pkt->data, (size_t)pkt->sz, 0); + capn_free(&cs); + + return pkt; +} + +static void test_config_push_reply_higher_term_adopts_and_steps_down(void) +{ + char tmp[256]; + strcpy(tmp, "/tmp/xce_cr22_term_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); + + cluster_t *c = t42_make_cluster(tmp); + TEST_ASSERT_NOT_NULL(c); + + atomic_store(&c->ha.role, CTRL_STANDBY); + atomic_store(&c->ha.term, 5); + ha_init_raft(&c->ha, c, /*n_peers*/3); + c->ha.raft.term = 5; + c->ha.raft.role = RAFT_LEADER; + + endpoint_t ep; + memset(&ep, 0, sizeof(ep)); + ep.owner = c; + + /* Reject (accepted=0) carrying a higher term (9 > 5): Req 7.4/7.5 + * require the term check to fire regardless of accept/reject -- a + * rejected push can still carry a useful higher term. */ + io_packet_t *pkt = + cpr22_build_config_push_reply_pkt(/*accepted*/0, + /*version_tag*/0xABCD, + /*term*/9); + TEST_ASSERT_NOT_NULL(pkt); + + int rc = myreq_handler_for_test(&ep, pkt); + TEST_ASSERT_EQUAL_INT(0, rc); + + TEST_ASSERT_EQUAL_UINT64_MESSAGE(9, c->ha.raft.term, + "raft-adapter-conformance 7.4: a config-push reply's higher term " + "must be adopted via raft_on_higher_term_observed"); + TEST_ASSERT_NOT_EQUAL_INT_MESSAGE(RAFT_LEADER, c->ha.raft.role, + "raft-adapter-conformance 7.5: leadership held at the lower term " + "must be stepped down once a higher term is observed"); + + /* NOTE: pkt is already fully freed by this point -- myreq_handler_for_test + * routes through myreq_handler -> xce_decode(), and xce_decode() + * unconditionally calls free_packet(pkt)+FREEUP(pkt) before returning + * (request.c). Do NOT free pkt here: it is a dangling pointer to freed + * memory and a second free_packet()/free() on it is a heap + * use-after-free/double-free (caught by ASAN regardless of + * detect_leaks). See myreq_handler_for_test's doc comment in cluster.c + * for the corrected ownership contract. */ + + t42_teardown(c); + char cmd[512]; + snprintf(cmd, sizeof(cmd), "rm -rf %s", tmp); + system(cmd); + printf("test_config_push_reply_higher_term_adopts_and_steps_down: PASS\n"); +} + #endif /* XCE_TESTING */ /* ── raw capnp shims (Task A) ───────────────────── */ @@ -6926,6 +7095,8 @@ void test_config_repl(void) RUN_TEST(test_configpush_transport_holds_endpoint_ref_across_send); /* raft-adapter-conformance task 2.1: config-push reject reply integrity */ RUN_TEST(test_config_push_sha256_mismatch_sends_explicit_reject_reply); + /* raft-adapter-conformance task 2.2: config-push reply term adoption */ + RUN_TEST(test_config_push_reply_higher_term_adopts_and_steps_down); /* Task 2.1: OPCode_walCatchupEntry handler */ RUN_TEST(test_wal_catchup_entry_advances_applied_lsn); RUN_TEST(test_wal_catchup_entry_error_skipped_no_crash);