lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(ha-bootstrap): wire OP_requestVote RPC through standard Request envelope

Browse source 
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.
This commit is contained in:
Rongsong Shen 2026-05-29 17:01:59 +00:00
parent e25d2cc645
commit 6753b364c8
4 changed files with 100 additions and 58 deletions
Download patch file Download diff file Expand all files Collapse all files

28
src/controller/cluster.c
View file

@ -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;

94
src/controller/ha.c
View file

@ -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;

34
src/include/typemap.h
View file

@ -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;

2
src/schema/controller.capnp
View file

@ -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;
}
}