diff --git a/src/controller/meson.build b/src/controller/meson.build index 6e00c6cf..c2e73746 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -151,6 +151,7 @@ controller_unit_test_src = [ 'testsuite/test_commit_resolve.c', 'testsuite/test_ha_vep_teardown.c', 'testsuite/test_ha_dispatch_integration.c', + 'testsuite/test_term_adoption_vectors.c', 'testsuite/test_ha_core_safety.c', 'testsuite/test_ha_partition.c', 'testsuite/test_waldump.c', diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 85fed323..24212b80 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -76,6 +76,7 @@ extern void test_commit_admission_pool(void); extern void test_commit_resolve(void); extern void test_ha_vep_teardown(void); extern void test_ha_dispatch_integration(void); +extern void test_term_adoption_vectors(void); extern void test_ha_core_safety(void); extern void test_ha_partition(void); extern void test_persist_snapshot_entities(void); @@ -237,6 +238,7 @@ int main(int argc, char* argv[]) TEST(test_commit_resolve); TEST(test_ha_vep_teardown); TEST(test_ha_dispatch_integration); + TEST(test_term_adoption_vectors); TEST(test_ha_core_safety); TEST(test_ha_partition); TEST(test_persist_snapshot_entities); diff --git a/src/controller/testsuite/test_term_adoption_vectors.c b/src/controller/testsuite/test_term_adoption_vectors.c new file mode 100644 index 00000000..2191e522 --- /dev/null +++ b/src/controller/testsuite/test_term_adoption_vectors.c @@ -0,0 +1,250 @@ +/* + * test_term_adoption_vectors.c — bug-1484 regression lock. + * + * bug-1421 added two term-discovery fallback vectors so a stale-term primary + * steps down instead of retrying AppendEntries against a follower that will + * reject forever: + * (a) heartbeat-reply state-report term adoption (ha.c ~6690-6711) + * (b) WAL-NACK reply_term adoption inside repl_ack_decode_and_dispatch + * (ha.c ~6805-6814) + * + * Neither vector had a test driving the real repl_ack_decode_and_dispatch + * path and asserting raft_on_higher_term_observed actually fires -- + * test_intake_higher_term_heartbeat_adopts_term (test_term_safety_adapter.c) + * looks like a match but drives a different code path (the AE-request + * UpdateTerm vector, not this AE-reply/state-report fallback). This file + * closes that gap, mirroring the pattern task 2.2's + * test_config_push_reply_higher_term_adopts_and_steps_down (test_config_repl.c) + * already established for the third (config-push) vector. + * + * Requirements: raft-adapter-conformance Req 7.5 (bug-1484 residual). + */ +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include + +#include + +#include "controller.capnp.h" +#include "ha.h" +#include "netio.h" +#include "xce/cluster_def.h" + +/* ── ha.c test seams ─────────────────────────────────────────────────────── */ + +extern int peer_ctrl_msg_handler_for_test(endpoint_t *ep, io_packet_t *pkt); + +extern endpoint_t *standby_ep_ctx_build_for_test(endpoint_t *ep, ha_state_t *ha, + cluster_t *c, int peer_idx); + +static ha_config_t g_cfg = { + .heartbeat_interval_ms = 3000, + .heartbeat_timeout_ms = 6000, + .election_timeout_ms = 500, + .probe_connect_timeout_ms = 50, + .max_promotion_retries = 3, + .election_jitter_max_ms = 0, + .election_priority_step_ms = 500, +}; + +/* Build Request{opcode=reply, body.replymsg=RequestReply{data_which= + * stateReportMsg, data.stateReportMsg=StateReport{appliedLsn,configTag,term}}} + * -- the exact wire shape ha_nop_reply_state_report populates in production + * (ha.c:3802-3811), wrapped by xce_reply() the same way cpr22's config-push + * reply builder mirrors handle_repl_config_push's send_config_push_reply. */ +static io_packet_t *build_state_report_pkt(uint64_t applied_lsn, + uint64_t config_tag, uint64_t term) +{ + struct capn cs; + struct capn_segment *seg; + struct StateReport sr_capn; + StateReport_ptr sr_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(&sr_capn, 0, sizeof(sr_capn)); + sr_capn.appliedLsn = applied_lsn; + sr_capn.configTag = config_tag; + sr_capn.term = term; + sr_ptr = new_StateReport(seg); + write_StateReport(&sr_capn, sr_ptr); + + memset(&rr_capn, 0, sizeof(rr_capn)); + rr_capn.seq = 1; + rr_capn.opcode = OPCode_nop; + rr_capn.status = 0; + rr_capn.data_which = RequestReply_data_stateReportMsg; + rr_capn.data.stateReportMsg = sr_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, "state-report pkt alloc"); + pkt->sz = capn_size(&cs); + TEST_ASSERT_TRUE_MESSAGE(alloc_packet(pkt) >= 0, "state-report pkt data alloc"); + capn_write_mem(&cs, (uint8_t *)pkt->data, (size_t)pkt->sz, 0); + capn_free(&cs); + + return pkt; +} + +/* Build Request{opcode=reply, body.replymsg=RequestReply{data_which= + * walackreplymsg, data.walackreplymsg=WalAckReply{term,lsn,success=0}}} -- + * a WAL catch-up entry NACK carrying a higher term. */ +static io_packet_t *build_wal_nack_pkt(uint64_t lsn, uint64_t term) +{ + struct capn cs; + struct capn_segment *seg; + struct WalAckReply wa_capn; + WalAckReply_ptr wa_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(&wa_capn, 0, sizeof(wa_capn)); + wa_capn.term = term; + wa_capn.lsn = lsn; + wa_capn.success = 0; + wa_ptr = new_WalAckReply(seg); + write_WalAckReply(&wa_capn, wa_ptr); + + memset(&rr_capn, 0, sizeof(rr_capn)); + rr_capn.seq = 1; + rr_capn.opcode = OPCode_nop; + rr_capn.status = 0; + rr_capn.data_which = RequestReply_data_walackreplymsg; + rr_capn.data.walackreplymsg = wa_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, "wal-nack pkt alloc"); + pkt->sz = capn_size(&cs); + TEST_ASSERT_TRUE_MESSAGE(alloc_packet(pkt) >= 0, "wal-nack pkt data alloc"); + capn_write_mem(&cs, (uint8_t *)pkt->data, (size_t)pkt->sz, 0); + capn_free(&cs); + + return pkt; +} + +/* Shared fixture: a CTRL_PRIMARY node holding leadership at term 5, with a + * promoted "heartbeat standby" peer endpoint (vctx->ctx == NULL, peer_idx + * absolute -- ha.c:9126's documented shape) routed via + * peer_ctrl_msg_handler_inline's CTRL_PRIMARY branch (repl_ack_decode_and_ + * dispatch). c is deliberately NULL: every c-dependent step downstream of + * the term check is guarded (`if (c) {...}`), so this drives the exact same + * term-adoption code the production wire path runs, without needing a live + * cluster_t/persist/replicator fixture. */ +static void tav_setup(ha_state_t *ha, endpoint_t *ep) +{ + TEST_ASSERT_EQUAL_INT(0, ha_init(ha, &g_cfg, 0)); + atomic_store_explicit(&ha->role, (int32_t)CTRL_PRIMARY, memory_order_release); + ha_init_raft(ha, NULL, 3); + ha->raft.term = 5; + ha->raft.role = RAFT_LEADER; + + memset(ep, 0, sizeof(*ep)); + standby_ep_ctx_build_for_test(ep, ha, NULL, 0); +} + +static void tav_teardown(ha_state_t *ha, endpoint_t *ep) +{ + free(ep->data); + ep->data = NULL; + ha_fini(ha); +} + +/* ── Vector (a): heartbeat-reply state-report term adoption ─────────────── */ + +static void test_heartbeat_state_report_higher_term_adopts_and_steps_down(void) +{ + ha_state_t ha; + endpoint_t ep; + tav_setup(&ha, &ep); + + io_packet_t *pkt = build_state_report_pkt(/*applied_lsn*/0, /*config_tag*/0, + /*term*/9); + TEST_ASSERT_NOT_NULL(pkt); + + int rc = peer_ctrl_msg_handler_for_test(&ep, pkt); + TEST_ASSERT_EQUAL_INT(0, rc); + + TEST_ASSERT_EQUAL_UINT64_MESSAGE(9, ha.raft.term, + "bug-1484/bug-1421 vector (a): a heartbeat-reply state report's higher " + "term must be adopted via raft_on_higher_term_observed through the " + "real repl_ack_decode_and_dispatch path"); + TEST_ASSERT_NOT_EQUAL_INT_MESSAGE(RAFT_LEADER, ha.raft.role, + "bug-1484/bug-1421 vector (a): leadership held at the lower term must " + "be stepped down once a higher term is observed via the heartbeat-" + "reply fallback"); + + tav_teardown(&ha, &ep); + printf("test_heartbeat_state_report_higher_term_adopts_and_steps_down: PASS\n"); +} + +/* ── Vector (b): WAL-NACK reply_term adoption ────────────────────────────── */ + +static void test_wal_nack_higher_term_adopts_and_steps_down(void) +{ + ha_state_t ha; + endpoint_t ep; + tav_setup(&ha, &ep); + + io_packet_t *pkt = build_wal_nack_pkt(/*lsn*/3, /*term*/9); + TEST_ASSERT_NOT_NULL(pkt); + + int rc = peer_ctrl_msg_handler_for_test(&ep, pkt); + TEST_ASSERT_EQUAL_INT(0, rc); + + TEST_ASSERT_EQUAL_UINT64_MESSAGE(9, ha.raft.term, + "bug-1484/bug-1421 vector (b): a WAL catch-up NACK's higher reply_term " + "must be adopted via raft_on_higher_term_observed through the real " + "repl_ack_decode_and_dispatch path"); + TEST_ASSERT_NOT_EQUAL_INT_MESSAGE(RAFT_LEADER, ha.raft.role, + "bug-1484/bug-1421 vector (b): leadership held at the lower term must " + "be stepped down once a higher term is observed via the WAL-NACK " + "fallback"); + + tav_teardown(&ha, &ep); + printf("test_wal_nack_higher_term_adopts_and_steps_down: PASS\n"); +} + +void test_term_adoption_vectors(void) +{ + UnitySetTestFile(__FILE__); + RUN_TEST(test_heartbeat_state_report_higher_term_adopts_and_steps_down); + RUN_TEST(test_wal_nack_higher_term_adopts_and_steps_down); +}