From 18c280429e05d61cee6d5a439c07a3a592b81e4f Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sat, 30 May 2026 08:09:33 +0000 Subject: [PATCH] feat(election-iomgr-refactor): add vote_ep_ctx_t and on_vote_connected MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - vote_ep_ctx_t: per-peer context (ha, c, ctx, peer_idx, term/tag/lsn/name) - election_timeout_ctx_t: cancelled flag + context for delay watchdog - on_vote_connected(): iomgr callback fires after SSL handshake; builds xce_request_t{opcode=OPCode_requestVote}, calls xce_request, sets msg_handler - vote_reply_msg_handler stub (returns 0; implemented in task 2.3) - Test seam: xce_vote_request_hook for unit testing without live SSL - 4 new unit tests: opcode, field encoding, msg_handler set, cancelled flag Context note: vctx->c is NULL in test helper — task 2.3 must use real cluster_t* --- src/controller/ha.c | 168 ++++++++++ src/controller/meson.build | 1 + src/controller/testsuite/controller_test.c | 2 + src/controller/testsuite/test_ha_vote_send.c | 329 +++++++++++++++++++ 4 files changed, 500 insertions(+) create mode 100644 src/controller/testsuite/test_ha_vote_send.c diff --git a/src/controller/ha.c b/src/controller/ha.c index 5c182506..10eca060 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -1016,6 +1016,174 @@ void handle_request_vote_reply(ha_state_t *ha, pthread_mutex_unlock(&ctx->lock); } +/* ── Task 2.2: iomgr-native vote send path ────────────────────────────────── + * + * vote_ep_ctx_t: per-peer context stored in ep->data for each outgoing vote + * endpoint. on_vote_connected fires after the TLS handshake completes (when + * endpoint_t.ssl_handshake_done transitions to 1 in io_handler). + * + * election_timeout_ctx_t: argument for the delay_task_ms watchdog that fires + * ha_elect_evaluate_and_signal when the election deadline expires without all + * peers replying. Uses an _Atomic int cancelled flag for lock-free cancellation. + */ + +typedef struct { + ha_state_t *ha; + cluster_t *c; + ha_elect_ctx_t *ctx; + int peer_idx; + uint64_t my_term; + uint64_t my_tag; + uint64_t my_lsn; + const char *my_name; +} vote_ep_ctx_t; + +typedef struct { + _Atomic int cancelled; + ha_state_t *ha; + cluster_t *c; + ha_elect_ctx_t *ctx; +} election_timeout_ctx_t; + +/* xce_request: defined in request.c, not yet in any public header */ +extern int xce_request(endpoint_t *ep, xce_request_t *req); + +/* + * vote_reply_msg_handler: forward declaration. + * Implemented in task 2.3. Registered on the endpoint by on_vote_connected + * so that the next iomgr EPOLLIN delivers the vote reply to this handler. + * Signature matches msg_handler_t = int (*)(void*) from netio.h. + */ +static int vote_reply_msg_handler(void *ep); + +#ifdef XCE_TESTING +/* + * xce_vote_request_hook: test-only seam. + * + * When non-NULL, on_vote_connected() calls this INSTEAD OF the real + * xce_request() so tests can capture the call without requiring a live + * iomgr or SSL endpoint. The hook receives the fully built + * xce_request_t and may inspect all fields. + * + * Pattern mirrors xce_retry_send_hook in request.c. + */ +int (*xce_vote_request_hook)(endpoint_t *ep, xce_request_t *req) = NULL; +#endif /* XCE_TESTING */ + +/* + * on_vote_connected: iomgr callback fired after TLS handshake completes on an + * outgoing vote endpoint. + * + * Steps: + * 1. Cast ep->data to vote_ep_ctx_t. + * 2. Build request_vote_req_t from ctx fields. + * 3. Build xce_request_t{opcode=OPCode_requestVote, body.requestvotemsg=&vrq}. + * 4. Call xce_vote_request_hook (test seam, XCE_TESTING only) — then + * call xce_request() to queue the packet on the iomgr send path. + * 5. Set ep->msg_handler = vote_reply_msg_handler so the next inbound + * packet is decoded as a RequestVoteReply. + * + * Req 2.3: send vote after TLS handshake. + * Req 1.2: vote RPC via iomgr path (xce_request enqueues to endpoint_enq). + */ +static void on_vote_connected(endpoint_t *ep) +{ + vote_ep_ctx_t *vctx = (vote_ep_ctx_t *)ep->data; + + /* Build the RequestVoteRequest payload */ + request_vote_req_t vrq; + memset(&vrq, 0, sizeof(vrq)); + vrq.term = vctx->my_term; + vrq.candidate_id = (char *)(vctx->my_name ? vctx->my_name : ""); + vrq.config_tag = vctx->my_tag; + vrq.last_log_lsn = vctx->my_lsn; + + /* Wrap in xce_request_t */ + xce_request_t req; + memset(&req, 0, sizeof(req)); + req.opcode = OPCode_requestVote; + req.body_which = Request_body_requestvotemsg; + req.body.requestvotemsg = &vrq; + +#ifdef XCE_TESTING + if (xce_vote_request_hook != NULL) + xce_vote_request_hook(ep, &req); + else + xce_request(ep, &req); +#else + xce_request(ep, &req); +#endif + + /* Switch to reply-reading mode: next inbound packet → vote_reply_msg_handler */ + ep->msg_handler = vote_reply_msg_handler; +} + +/* + * vote_reply_msg_handler: stub — implemented in task 2.3. + * Declared here so on_vote_connected can reference it. + * Signature matches msg_handler_t = int (*)(void*) from netio.h. + */ +static int vote_reply_msg_handler(void *ep) +{ + (void)ep; + /* Task 2.3 will implement this; stub returns 0 to avoid crash if called */ + return 0; +} + +#ifdef XCE_TESTING +/* + * vote_ep_ctx_build_for_test: test helper that allocates a vote_ep_ctx_t, + * fills it from the supplied parameters, and wires up ep->data and + * ep->on_connected so the test can drive on_vote_connected indirectly. + * + * Returns ep (same pointer passed in) for call-chain convenience. + * Caller is responsible for free(ep->data) after use. + */ +endpoint_t *vote_ep_ctx_build_for_test(endpoint_t *ep, + ha_state_t *ha, + ha_elect_ctx_t *ctx, + int peer_idx, + uint64_t my_term, + uint64_t my_tag, + uint64_t my_lsn, + const char *my_name) +{ + vote_ep_ctx_t *vctx = calloc(1, sizeof(*vctx)); + if (!vctx) return ep; /* test will fail when on_connected is NULL */ + + vctx->ha = ha; + vctx->c = NULL; + vctx->ctx = ctx; + vctx->peer_idx = peer_idx; + vctx->my_term = my_term; + vctx->my_tag = my_tag; + vctx->my_lsn = my_lsn; + vctx->my_name = my_name; + + ep->data = vctx; + ep->on_connected = on_vote_connected; + return ep; +} + +/* + * election_timeout_fn_for_test: test helper that exercises the election timeout + * callback logic with an explicit cancelled_val instead of an atomic field. + * Allows testing the cancelled-gate without setting up a real delaymgr. + * + * Returns 0 in all cases (mirrors the real callback's return value). + */ +int election_timeout_fn_for_test(ha_state_t *ha, ha_elect_ctx_t *ctx, + int cancelled_val) +{ + if (cancelled_val) + return 0; + pthread_mutex_lock(&ctx->lock); + ha_elect_evaluate_and_signal(ha, ctx); + pthread_mutex_unlock(&ctx->lock); + return 0; +} +#endif /* XCE_TESTING */ + /* ── Per-peer vote-request thread ────────────────────────────────────────── * * vote_thread_ctx_t: arguments passed to each vote_request_thread(). diff --git a/src/controller/meson.build b/src/controller/meson.build index 41765baf..c20444ed 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -111,6 +111,7 @@ controller_unit_test_src = [ 'testsuite/test_ha_connect.c', 'testsuite/test_ha_ctrlhello.c', 'testsuite/test_ha_failover.c', + 'testsuite/test_ha_vote_send.c', 'startup_validation.c', 'worker.c', 'persist.c', diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index ff7fb1c1..5cfd8ed8 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -43,6 +43,7 @@ extern void test_persist_truncate(void); extern void test_ha_connect(void); extern void test_ha_ctrlhello(void); extern void test_ha_failover(void); +extern void test_ha_vote_send(void); void setUp(void) { return; } @@ -155,6 +156,7 @@ int main(int argc, char* argv[]) TEST(test_ha_connect); TEST(test_ha_ctrlhello); TEST(test_ha_failover); + TEST(test_ha_vote_send); return UNITY_END(); } diff --git a/src/controller/testsuite/test_ha_vote_send.c b/src/controller/testsuite/test_ha_vote_send.c new file mode 100644 index 00000000..2c2de1a5 --- /dev/null +++ b/src/controller/testsuite/test_ha_vote_send.c @@ -0,0 +1,329 @@ +/* + * test_ha_vote_send.c — Unity tests for task 2.2: vote_ep_ctx_t and + * on_vote_connected callback (election-iomgr-refactor spec). + * + * Observable: unit test with mock endpoint verifies xce_request called with + * correct opcode and fields after on_vote_connected fires. + * + * Strategy: XCE_TESTING test seam — xce_vote_request_hook captures the call + * to xce_request from on_vote_connected, letting us inspect the xce_request_t + * without touching the real iomgr/endpoint_enq path. + * + * Req coverage: 2.3 (send vote after handshake), 1.2 (vote RPC via iomgr path) + */ +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include + +#include + +#include "ha.h" +#include "netio.h" +#include "controller.capnp.h" + +/* ── Forward declarations for functions under test ───────────────────────── + * + * vote_ep_ctx_t, on_vote_connected, election_timeout_ctx_t are defined in + * ha.c (file-scoped under XCE_TESTING preprocessor for the hook; the structs + * and callback are static but the hook pointer is exposed). + * + * on_vote_connected is static — we call it indirectly by setting ep->on_connected + * and invoking it through the endpoint pointer, but since we can't extern a static + * function, we test via the hook seam exclusively: set xce_vote_request_hook and + * invoke on_vote_connected by address via the function pointer stored in ep->on_connected. + * + * vote_ep_ctx_t_build: test helper exposed only under XCE_TESTING that allocates + * and fills a vote_ep_ctx_t, sets ep->on_connected, ep->data, so the test can + * call ep->on_connected(ep) to drive on_vote_connected. + */ + +/* Test seam: pointer is declared in ha.c under XCE_TESTING and exposed here */ +extern int (*xce_vote_request_hook)(endpoint_t *ep, xce_request_t *req); + +/* vote_ep_ctx_build: test helper exposed from ha.c under XCE_TESTING. + * Allocates a vote_ep_ctx_t, fills it from the parameters, sets ep->data and + * ep->on_connected. Caller owns the returned pointer. */ +typedef struct vote_ep_ctx_s vote_ep_ctx_t; /* defined in ha.c — opaque here */ + +extern endpoint_t *vote_ep_ctx_build_for_test(endpoint_t *ep, + ha_state_t *ha, + ha_elect_ctx_t *ctx, + int peer_idx, + uint64_t my_term, + uint64_t my_tag, + uint64_t my_lsn, + const char *my_name); + +/* ── Minimal ha_config_t for tests ─────────────────────────────────────── */ + +static ha_config_t g_vote_cfg = { + .heartbeat_interval_ms = 3000, + .heartbeat_timeout_ms = 10000, + .election_timeout_ms = 2000, + .probe_connect_timeout_ms = 500, + .max_promotion_retries = 5, + .election_jitter_max_ms = 0, + .election_priority_step_ms = 500, +}; + +/* ── Mock xce_request hook ──────────────────────────────────────────────── */ + +static int g_hook_call_count = 0; +static endpoint_t *g_hook_ep = NULL; +static uint32_t g_hook_opcode = 0; +static int g_hook_body_which = 0; +static uint64_t g_captured_term = 0; +static uint64_t g_captured_tag = 0; +static uint64_t g_captured_lsn = 0; +static char g_captured_name[256] = {0}; + +static int mock_vote_request_hook(endpoint_t *ep, xce_request_t *req) +{ + g_hook_call_count++; + g_hook_ep = ep; + g_hook_opcode = req->opcode; + g_hook_body_which = req->body_which; + + if (req->body_which == Request_body_requestvotemsg && + req->body.requestvotemsg != NULL) { + g_captured_term = req->body.requestvotemsg->term; + g_captured_tag = req->body.requestvotemsg->config_tag; + g_captured_lsn = req->body.requestvotemsg->last_log_lsn; + if (req->body.requestvotemsg->candidate_id) + strncpy(g_captured_name, req->body.requestvotemsg->candidate_id, + sizeof(g_captured_name) - 1); + } + return 0; +} + +static void reset_mock(void) +{ + g_hook_call_count = 0; + g_hook_ep = NULL; + g_hook_opcode = 0; + g_hook_body_which = 0; + g_captured_term = 0; + g_captured_tag = 0; + g_captured_lsn = 0; + memset(g_captured_name, 0, sizeof(g_captured_name)); + xce_vote_request_hook = mock_vote_request_hook; +} + +/* ── Helper: make a minimal zeroed endpoint_t on the stack ─────────────── */ + +static void make_endpoint(endpoint_t *ep) +{ + memset(ep, 0, sizeof(*ep)); + ep->fd = -1; +} + +/* ── Helper: allocate a minimal ha_elect_ctx_t ─────────────────────────── */ + +static ha_elect_ctx_t *alloc_ctx_for_vote(int n_peers) +{ + ha_elect_ctx_t *ctx = calloc(1, sizeof(*ctx) + + (size_t)n_peers * sizeof(ctx->votes[0])); + if (!ctx) { fprintf(stderr, "alloc_ctx_for_vote: calloc failed\n"); exit(1); } + ctx->n_peers = n_peers; + ctx->votes_granted = 1; /* self-vote */ + sem_init(&ctx->done, 0, 0); + return ctx; +} + +static void free_ctx_for_vote(ha_elect_ctx_t *ctx) +{ + sem_destroy(&ctx->done); + free(ctx); +} + +/* ── Test: on_vote_connected calls hook with OPCode_requestVote ─────────── */ + +static void test_on_vote_connected_calls_hook_with_correct_opcode(void) +{ + reset_mock(); + + ha_state_t ha; + ha_init(&ha, &g_vote_cfg, 0); + atomic_store_explicit(&ha.term, 3ULL, memory_order_release); + + ha_elect_ctx_t *ctx = alloc_ctx_for_vote(2); + + endpoint_t ep; + make_endpoint(&ep); + + /* vote_ep_ctx_build_for_test sets ep->data and ep->on_connected */ + vote_ep_ctx_build_for_test(&ep, &ha, ctx, + 0, /* peer_idx */ + 3ULL, /* my_term */ + 100ULL, /* my_tag */ + 50ULL, /* my_lsn */ + "ctrl-a"); + + TEST_ASSERT_NOT_NULL_MESSAGE(ep.on_connected, + "ep->on_connected must be set by vote_ep_ctx_build_for_test"); + + /* Fire on_vote_connected */ + ep.on_connected(&ep); + + /* Verify hook was called exactly once */ + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_hook_call_count, + "xce_vote_request_hook must be called exactly once by on_vote_connected"); + + /* Verify opcode */ + TEST_ASSERT_EQUAL_UINT32_MESSAGE(OPCode_requestVote, g_hook_opcode, + "xce_request must be called with opcode OPCode_requestVote"); + + /* Verify body_which */ + TEST_ASSERT_EQUAL_INT_MESSAGE(Request_body_requestvotemsg, g_hook_body_which, + "body_which must be Request_body_requestvotemsg"); + + /* Verify endpoint pointer passed through */ + TEST_ASSERT_EQUAL_PTR_MESSAGE(&ep, g_hook_ep, + "hook must receive the same endpoint pointer that on_vote_connected received"); + + /* Cleanup: ep.data was allocated by vote_ep_ctx_build_for_test; free it */ + free(ep.data); + free_ctx_for_vote(ctx); + ha_fini(&ha); + printf("test_on_vote_connected_calls_hook_with_correct_opcode: PASS\n"); +} + +/* ── Test: on_vote_connected encodes correct term, tag, lsn, name ────────── */ + +static void test_on_vote_connected_encodes_correct_fields(void) +{ + reset_mock(); + + ha_state_t ha; + ha_init(&ha, &g_vote_cfg, 1); + atomic_store_explicit(&ha.term, 7ULL, memory_order_release); + + ha_elect_ctx_t *ctx = alloc_ctx_for_vote(2); + + endpoint_t ep; + make_endpoint(&ep); + + vote_ep_ctx_build_for_test(&ep, &ha, ctx, + 1, /* peer_idx */ + 7ULL, /* my_term */ + 200ULL, /* my_tag */ + 99ULL, /* my_lsn */ + "ctrl-b"); + + ep.on_connected(&ep); + + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_hook_call_count, + "hook must be called exactly once"); + + TEST_ASSERT_EQUAL_UINT64_MESSAGE(7ULL, g_captured_term, + "request_vote_req_t.term must match my_term from vote_ep_ctx_t"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(200ULL, g_captured_tag, + "request_vote_req_t.config_tag must match my_tag from vote_ep_ctx_t"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(99ULL, g_captured_lsn, + "request_vote_req_t.last_log_lsn must match my_lsn from vote_ep_ctx_t"); + TEST_ASSERT_EQUAL_STRING_MESSAGE("ctrl-b", g_captured_name, + "request_vote_req_t.candidate_id must match my_name from vote_ep_ctx_t"); + + free(ep.data); + free_ctx_for_vote(ctx); + ha_fini(&ha); + printf("test_on_vote_connected_encodes_correct_fields: PASS\n"); +} + +/* ── Test: on_vote_connected sets ep->msg_handler after sending ─────────── */ + +static void test_on_vote_connected_sets_msg_handler(void) +{ + reset_mock(); + + ha_state_t ha; + ha_init(&ha, &g_vote_cfg, 0); + atomic_store_explicit(&ha.term, 1ULL, memory_order_release); + + ha_elect_ctx_t *ctx = alloc_ctx_for_vote(2); + + endpoint_t ep; + make_endpoint(&ep); + + vote_ep_ctx_build_for_test(&ep, &ha, ctx, + 0, + 1ULL, + 100ULL, + 50ULL, + "ctrl-a"); + + /* msg_handler starts NULL */ + TEST_ASSERT_NULL_MESSAGE(ep.msg_handler, + "msg_handler must be NULL before on_vote_connected fires"); + + ep.on_connected(&ep); + + /* After on_vote_connected fires, ep->msg_handler must be set to + * vote_reply_msg_handler (forward-declared in 2.3; non-NULL here) */ + TEST_ASSERT_NOT_NULL_MESSAGE(ep.msg_handler, + "on_vote_connected must set ep->msg_handler to vote_reply_msg_handler"); + + free(ep.data); + free_ctx_for_vote(ctx); + ha_fini(&ha); + printf("test_on_vote_connected_sets_msg_handler: PASS\n"); +} + +/* ── Test: election_timeout_ctx_t cancelled flag prevents action ─────────── */ + +static void test_election_timeout_ctx_cancelled_flag(void) +{ + /* election_timeout_ctx_t carries _Atomic int cancelled. + * When cancelled == 1, the delaymgr callback must be a no-op. + * We test this by directly calling the exported helper + * election_timeout_fn_for_test (exposed under XCE_TESTING). */ + extern int election_timeout_fn_for_test(ha_state_t *ha, ha_elect_ctx_t *ctx, + int cancelled_val); + + ha_state_t ha; + ha_init(&ha, &g_vote_cfg, 0); + atomic_store_explicit(&ha.term, 1ULL, memory_order_release); + + ha_elect_ctx_t *ctx = alloc_ctx_for_vote(2); + /* Set up so if evaluate fires, we'd know (done posted changes result) */ + ctx->votes_granted = 1; /* below quorum — would yield STANDBY */ + ctx->n_replied = 2; + + /* cancelled=1 → no evaluate → done NOT posted */ + int rc = election_timeout_fn_for_test(&ha, ctx, /*cancelled=*/1); + TEST_ASSERT_EQUAL_INT(0, rc); /* must return 0 (no-op) */ + + int try_rc = sem_trywait(&ctx->done); + TEST_ASSERT_EQUAL_INT_MESSAGE(-1, try_rc, + "cancelled election_timeout must NOT post ctx->done"); + TEST_ASSERT_EQUAL_INT(EAGAIN, errno); + + /* cancelled=0 → evaluate runs → done posted */ + ctx->done_posted = 0; /* reset guard */ + ctx->votes[0].my_tag = 100ULL; + ctx->votes[0].my_lsn = 50ULL; + rc = election_timeout_fn_for_test(&ha, ctx, /*cancelled=*/0); + TEST_ASSERT_EQUAL_INT(0, rc); + + try_rc = sem_trywait(&ctx->done); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, try_rc, + "not-cancelled election_timeout must post ctx->done"); + + free_ctx_for_vote(ctx); + ha_fini(&ha); + printf("test_election_timeout_ctx_cancelled_flag: PASS\n"); +} + +/* ── Suite entry point ───────────────────────────────────────────────────── */ + +void test_ha_vote_send(void) +{ + RUN_TEST(test_on_vote_connected_calls_hook_with_correct_opcode); + RUN_TEST(test_on_vote_connected_encodes_correct_fields); + RUN_TEST(test_on_vote_connected_sets_msg_handler); + RUN_TEST(test_election_timeout_ctx_cancelled_flag); +}