From 6753b364c804e32f9f1e335b9281e78df691c85d Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Fri, 29 May 2026 17:01:59 +0000 Subject: [PATCH] fix(ha-bootstrap): wire OP_requestVote RPC through standard Request envelope MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The vote_request_thread was sending raw capnp(RequestVoteRequest) bytes but the server-side iomgr decodes every incoming packet as a Request struct. This caused the term field (uint64) to be misread as the opcode field (uint16), so vote RPCs were silently misrouted (term=1 → OPCode_reply, term=2 → OPCode_hello) and no node ever got a vote from its peers. Fix: - controller.capnp: add requestvotemsg @27 to Request.body and requestvotereplymsg @14 to RequestReply.data - typemap.h: move request_vote_req_t/reply_t before xce_request_t; add requestvotemsg/requestvotereplymsg to body/data unions - ha.c: rewrite vote_request_thread send/recv to use xce_encode() wrapper and decode outer Request→inner RequestReply on receive - cluster.c: add case OPCode_requestVote handler calling handle_request_vote() and replying via xce_reply() All 188 unit tests pass. --- src/controller/cluster.c | 28 +++++++++++ src/controller/ha.c | 94 ++++++++++++++++++++----------------- src/include/typemap.h | 34 ++++++++------ src/schema/controller.capnp | 2 + 4 files changed, 100 insertions(+), 58 deletions(-) diff --git a/src/controller/cluster.c b/src/controller/cluster.c index 6799ff1e..3a1b70f0 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -281,6 +281,34 @@ static int myreq_handler(vtask_t* task) case OPCode_clrpreason: rc = do_clr_preason(c, ep, &request); break; + case OPCode_requestVote: { + request_vote_req_t *vrq = request.body.requestvotemsg; + uint64_t cand_tag = vrq ? vrq->config_tag : 0; + uint64_t cand_lsn = vrq ? vrq->last_log_lsn : 0; + uint64_t cand_term = vrq ? vrq->term : 0; + uint64_t my_tag = c->myconf->config_version_tag; + uint64_t my_lsn = (uint64_t)atomic_load_explicit( + &c->ha.applied_lsn, memory_order_acquire); + int granted = handle_request_vote( + &c->ha, cand_term, vrq ? vrq->candidate_id : "", + cand_tag, cand_lsn, my_tag, my_lsn, + 0 /* in_catch_up_flag: not implemented here */); + + request_vote_reply_t vr_rep; + memset(&vr_rep, 0, sizeof(vr_rep)); + vr_rep.term = (uint64_t)atomic_load_explicit( + &c->ha.term, memory_order_acquire); + vr_rep.vote_granted = (granted > 0) ? 1 : 0; + vr_rep.peer_config_tag = my_tag; + vr_rep.peer_last_log_lsn = my_lsn; + + reply.status = 0; + reply.data_which = RequestReply_data_requestvotereplymsg; + reply.data.requestvotereplymsg = &vr_rep; + xce_reply(ep, c->myhost, &reply); + rc = 0; + break; + } case OPCode_replConfig: rc = handle_config_request(c, ep, &request); break; diff --git a/src/controller/ha.c b/src/controller/ha.c index ea0043f5..53af8bc4 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -28,6 +28,7 @@ #include "xce/errnox.h" #include "controller.capnp.h" #include "persist.h" +#include "request.h" #include "utils.h" /* ── HA dispatch gate error codes (defined here; declared extern in errnox.h) ── */ @@ -1091,50 +1092,43 @@ static void *vote_request_thread(void *arg) if (sr <= 0) goto deny; } - /* ── 3. Encode + send RequestVoteRequest ─────────────── */ + /* ── 3. Encode + send RequestVoteRequest wrapped in a Request envelope ── */ { - struct capn c; - struct capn_segment *cs; - struct RequestVoteRequest req_s; - RequestVoteRequest_ptr req_p; - uint8_t *buf = NULL; - int64_t blen = 0; + request_vote_req_t vote_req; + memset(&vote_req, 0, sizeof(vote_req)); + vote_req.term = vctx->my_term; + vote_req.candidate_id = (char *)""; /* anonymous; peer reads configTag/lsn */ + vote_req.config_tag = vctx->my_tag; + vote_req.last_log_lsn = vctx->my_lsn; - capn_init_malloc(&c); - cs = capn_root(&c).seg; - memset(&req_s, 0, sizeof(req_s)); - req_s.term = vctx->my_term; - req_s.candidateId.str = ""; req_s.candidateId.len = 0; /* no name needed here */ - req_s.configTag = vctx->my_tag; - req_s.lastLogLsn = vctx->my_lsn; + xce_request_t xreq; + memset(&xreq, 0, sizeof(xreq)); + xreq.opcode = OPCode_requestVote; + xreq.body_which = Request_body_requestvotemsg; + xreq.body.requestvotemsg = &vote_req; - req_p = new_RequestVoteRequest(cs); - write_RequestVoteRequest(&req_s, req_p); - capn_setp(capn_root(&c), 0, req_p.p); + io_packet_t pkt; + memset(&pkt, 0, sizeof(pkt)); + if (xce_encode(&xreq, &pkt) < 0) + goto deny; - blen = capn_size(&c); - if (blen > 0) buf = malloc((size_t)blen); - if (!buf || capn_write_mem(&c, buf, (size_t)blen, 0) < 0) { - free(buf); capn_free(&c); goto deny; - } - capn_free(&c); - - uint64_t hdr = (uint64_t)blen; + uint64_t hdr = (uint64_t)pkt.sz; + size_t body_len = pkt.sz; int nsent; if (ssl) { nsent = SSL_write(ssl, &hdr, (int)sizeof(hdr)); if (nsent == (int)sizeof(hdr)) - nsent = SSL_write(ssl, buf, (int)blen); + nsent = SSL_write(ssl, pkt.data, (int)body_len); } else { nsent = (int)write(sock, &hdr, sizeof(hdr)); if (nsent == (int)sizeof(hdr)) - nsent = (int)write(sock, buf, (size_t)blen); + nsent = (int)write(sock, pkt.data, body_len); } - free(buf); - if (nsent != (int)blen) goto deny; + free_packet(&pkt); + if (nsent != (int)body_len) goto deny; } - /* ── 4. Read + decode RequestVoteReply ───────────────── */ + /* ── 4. Read + decode RequestReply containing RequestVoteReply ─────── */ { uint64_t hdr = 0; size_t off = 0; @@ -1166,28 +1160,42 @@ static void *vote_request_thread(void *arg) off += (size_t)n; } - struct capn c; - struct RequestVoteReply rep_s; - RequestVoteReply_ptr rep_p; + /* Unwrap outer Request{opcode=reply, body.replymsg=RequestReply{...}} */ + struct capn capn_ctx; + struct Request outer_rq; + Request_ptr outer_rp; + struct RequestReply rep_s; - if (capn_init_mem(&c, buf, (size_t)hdr, 0) != 0) { + if (capn_init_mem(&capn_ctx, buf, (size_t)hdr, 0) != 0) { free(buf); goto deny; } - rep_p.p = capn_getp(capn_root(&c), 0, 1); - read_RequestVoteReply(&rep_s, rep_p); - capn_free(&c); + outer_rp.p = capn_getp(capn_root(&capn_ctx), 0, 1); + read_Request(&outer_rq, outer_rp); + + if (outer_rq.body_which != Request_body_replymsg || + !outer_rq.body.replymsg.p.data) { + capn_free(&capn_ctx); free(buf); goto deny; + } + read_RequestReply(&rep_s, outer_rq.body.replymsg); + capn_free(&capn_ctx); free(buf); + if (rep_s.data_which != RequestReply_data_requestvotereplymsg) + goto deny; + + struct RequestVoteReply vr_raw; + read_RequestVoteReply(&vr_raw, rep_s.data.requestvotereplymsg); + /* Fencing: if peer's term > ours, step down immediately */ - if (rep_s.term > vctx->my_term) - atomic_store_explicit(&ha->term, rep_s.term, memory_order_release); + if (vr_raw.term > vctx->my_term) + atomic_store_explicit(&ha->term, vr_raw.term, memory_order_release); /* Record the reply */ handle_request_vote_reply(ha, ctx, idx, - rep_s.term, - rep_s.voteGranted ? 1 : 0, - rep_s.peerConfigTag, - rep_s.peerLastLogLsn); + vr_raw.term, + vr_raw.voteGranted ? 1 : 0, + vr_raw.peerConfigTag, + vr_raw.peerLastLogLsn); } goto cleanup; diff --git a/src/include/typemap.h b/src/include/typemap.h index ca333f20..dfe1744f 100644 --- a/src/include/typemap.h +++ b/src/include/typemap.h @@ -172,6 +172,20 @@ typedef struct { } data; } xce_update_data_t; +typedef struct { + uint64_t term; + char *candidate_id; + uint64_t config_tag; + uint64_t last_log_lsn; +} request_vote_req_t; + +typedef struct { + uint64_t term; + int vote_granted; + uint64_t peer_config_tag; + uint64_t peer_last_log_lsn; +} request_vote_reply_t; + typedef struct { uint64_t seq; uint32_t opcode; @@ -186,6 +200,7 @@ typedef struct { repl_config_push_reply_t* replconfigpushreplymsg; upgrade_config_reply_t* upgradeconfigreplymsg; ctrl_hello_reply_t* ctrlhelloreplymsg; + request_vote_reply_t* requestvotereplymsg; } data; /* Req 7: standby redirect fields — populated when status == XCE_ERR_STANDBY */ char *redirect_host; @@ -221,6 +236,7 @@ typedef struct { upgrade_config_req_t* upgradeconfigrequestmsg; heartbeat_msg_t* heartbeatmsg; ctrl_hello_req_t* ctrlhellomsg; + request_vote_req_t* requestvotemsg; void* walcatchupentrymsg; /* wal_catchup_entry_t* */ void* snapshottransfermsg; /* snapshot_transfer_t* */ } body; @@ -229,21 +245,9 @@ typedef struct { /* ── HA election and replication Cap'n Proto mapped types ──────────────── * C-side counterparts for the HA protocol capnp structs added in the * ha-config-bootstrap spec (controller.capnp structs with $C.mapname). - * Must be declared here so controller.capnp.h can reference them. */ - -typedef struct { - uint64_t term; - char *candidate_id; - uint64_t config_tag; - uint64_t last_log_lsn; -} request_vote_req_t; - -typedef struct { - uint64_t term; - int vote_granted; - uint64_t peer_config_tag; - uint64_t peer_last_log_lsn; -} request_vote_reply_t; + * Must be declared here so controller.capnp.h can reference them. + * request_vote_req_t and request_vote_reply_t are defined above + * xce_reply_t so they can be used in the xce_request_t body union. */ typedef struct { char *name; diff --git a/src/schema/controller.capnp b/src/schema/controller.capnp index 4b199505..9376325f 100644 --- a/src/schema/controller.capnp +++ b/src/schema/controller.capnp @@ -558,6 +558,7 @@ struct RequestReply $C.mapname("xce_reply_t"){ replconfigpushreplymsg @9: ReplConfigPushReply; upgradeconfigreplymsg @10: UpgradeConfigReply; ctrlhelloreplymsg @13: CtrlHelloReply; + requestvotereplymsg @14: RequestVoteReply; } redirectHost @11: Text $C.mapname("redirect_host"); @@ -623,6 +624,7 @@ struct Request $C.mapname("xce_request_t"){ upgradeconfigrequestmsg @24: UpgradeConfigRequest; heartbeatmsg @25: HeartbeatMsg; ctrlhellomsg @26: CtrlHelloRequest; + requestvotemsg @27: RequestVoteRequest; } }