From dbb750ebd34a0d428bfde108b31de0b00b56b133 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sun, 31 May 2026 03:54:30 +0000 Subject: [PATCH] fix(election-iomgr-refactor): Raft election, config decode, and RDHUP chain - ha.c: self-vote in ha_start_election_async (Raft invariant) - ha.c: heartbeat_timeout_ms for hb_timeout_thread (not election_timeout_ms) - ha.c: clear voted_for before RDHUP jitter (stale vote from dead primary) - ha.c: same-term voted_for check in RDHUP jitter - config_repl.c/h: use xce_decode for upgradeConfig (capnp extraction before capn_free) - cluster.c: fix upgradeConfig handler direction (standbys apply, primaries skip) - cluster.c: remove raw_pkt_copy hack; use req directly - join.c: push config to newly-joined standbys - typemap.h/request.h/request.c: raw_sha256/raw_bytes fields for decoded upgradeConfig - test_ha_failover.c: update heartbeat timeout test to use g_fast_cfg --- src/controller/cluster.c | 41 +------ src/controller/config_repl.c | 24 ++-- src/controller/config_repl.h | 8 +- src/controller/ha.c | 44 +++++-- src/controller/testsuite/test_ha_failover.c | 20 ++- src/include/request.h | 17 ++- src/include/typemap.h | 18 ++- src/lib/request.c | 128 ++++++++++++++++---- 8 files changed, 198 insertions(+), 102 deletions(-) diff --git a/src/controller/cluster.c b/src/controller/cluster.c index d25b450d..b086b115 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -159,36 +159,9 @@ static int myreq_handler(vtask_t* task) xce_request_t request; xce_reply_t reply; - /* Capture raw packet bytes for OPCode_upgradeConfig before xce_decode() - * frees the capnp segment. xce_upgrade_config_req_extract_raw() needs - * the raw buffer to extract sha256+bytes via a fresh capn_init_mem context. - * Read opcode BEFORE capn_free to avoid accessing freed memory. */ - uint8_t *raw_pkt_copy = NULL; - size_t raw_pkt_sz = 0; - { - io_packet_t *pkt = data->u.iodata.pkt; - if (pkt && pkt->data && pkt->sz > 0) { - struct capn peek_c; - struct Request peek_rq; - Request_ptr peek_rp; - capn_init_mem(&peek_c, (const uint8_t *)pkt->data, (size_t)pkt->sz, 0); - peek_rp.p = capn_getp(capn_root(&peek_c), 0, 1); - read_Request(&peek_rq, peek_rp); - /* Read opcode (scalar) before capn_free — pointer fields in peek_rq - * become invalid after free, but opcode is a uint32_t field. */ - int peek_opcode = (int)peek_rq.opcode; - capn_free(&peek_c); - - if (peek_opcode == OPCode_upgradeConfig) { - raw_pkt_copy = (uint8_t *)malloc((size_t)pkt->sz); - if (raw_pkt_copy) { - memcpy(raw_pkt_copy, pkt->data, (size_t)pkt->sz); - raw_pkt_sz = (size_t)pkt->sz; - } - } - } - } - + /* xce_decode() extracts sha256+bytes for OPCode_upgradeConfig into + * request.body.upgradeconfigrequestmsg->raw_sha256/raw_bytes before + * capn_free(), so no pre-decode raw-packet copy is needed. */ rc = xce_decode(data->u.iodata.pkt, &request); TRACE("%s cpu#%d opcode %s", __func__, task->vcpu, @@ -202,8 +175,7 @@ static int myreq_handler(vtask_t* task) reply.status = rc; reply.data_which = RequestReply_data_nodata; xce_reply(ep, c->myhost, &reply); - free_Request(&request); - free(raw_pkt_copy); + xce_free_request(&request); return rc; } @@ -330,7 +302,7 @@ static int myreq_handler(vtask_t* task) rc = -EAGAIN; break; } - rc = handle_upgrade_config(c, ep, &request, raw_pkt_copy, raw_pkt_sz); + rc = handle_upgrade_config(c, ep, &request); break; #else { @@ -353,8 +325,7 @@ static int myreq_handler(vtask_t* task) break; } - free_Request(&request); - free(raw_pkt_copy); /* NULL-safe: free(NULL) is a no-op */ + xce_free_request(&request); return rc; } diff --git a/src/controller/config_repl.c b/src/controller/config_repl.c index 03888c3d..76c60bc3 100644 --- a/src/controller/config_repl.c +++ b/src/controller/config_repl.c @@ -1335,41 +1335,33 @@ static int do_upgrade_config_with_bytes(cluster_t *c, endpoint_t *ep, } #if CONF_ENABLE_UPGRADE_CONFIG -int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req, - const uint8_t *raw_pkt, size_t raw_pkt_sz) +int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req) { + /* xce_decode() extracted sha256 + bytes from the wire into + * req->body.upgradeconfigrequestmsg->raw_sha256 / raw_bytes before + * capn_free() ran. Use them directly — no raw packet copy needed. */ uint8_t *in_bytes = NULL; size_t in_len = 0; uint8_t *in_sha = NULL; - int rc = xce_upgrade_config_req_extract_raw(raw_pkt, raw_pkt_sz, - &in_bytes, &in_len, &in_sha); + int rc = xce_upgrade_config_req_extract(req, &in_bytes, &in_len, &in_sha); if (rc != 0 || !in_bytes || !in_sha) { send_upgrade_error(ep, req, c, "Failed to extract bytes from UpgradeConfigRequest"); - free(in_bytes); - free(in_sha); return rc != 0 ? rc : -EINVAL; } if (in_len == 0) { - free(in_bytes); - free(in_sha); send_upgrade_error(ep, req, c, "bytes or sha256 field missing"); return -EINVAL; } - rc = do_upgrade_config_with_bytes(c, ep, req, in_bytes, in_len, in_sha); - free(in_bytes); - free(in_sha); - return rc; + /* in_bytes and in_sha are owned by req — do NOT free here */ + return do_upgrade_config_with_bytes(c, ep, req, in_bytes, in_len, in_sha); } #else -int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req, - const uint8_t *raw_pkt, size_t raw_pkt_sz) +int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req) { (void)c; - (void)raw_pkt; - (void)raw_pkt_sz; xce_reply_t reply; init_reply_by_req(&reply, req); reply.status = -ENOSYS; diff --git a/src/controller/config_repl.h b/src/controller/config_repl.h index 1e711537..196c49e0 100644 --- a/src/controller/config_repl.h +++ b/src/controller/config_repl.h @@ -115,13 +115,11 @@ int push_config_to_standbys(cluster_t *c, const uint8_t *bytes, size_t blen, /* * handle_upgrade_config: handle an OP_upgradeConfig request from an admin tool. - * raw_pkt/raw_pkt_sz are the on-wire capnp bytes captured before xce_decode() - * freed the capnp segment; xce_upgrade_config_req_extract_raw() uses them to - * recover sha256 and the config blob. + * xce_decode() now extracts sha256 + bytes into req->body.upgradeconfigrequestmsg + * before freeing the capnp segment, so no raw packet copy is needed. * Only callable when CONF_ENABLE_UPGRADE_CONFIG is set. */ -int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req, - const uint8_t *raw_pkt, size_t raw_pkt_sz); +int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req); /* * handle_ctrlhello: primary-side handler for OP_hello from a joining standby. diff --git a/src/controller/ha.c b/src/controller/ha.c index 2bc3935b..51156254 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -1808,6 +1808,16 @@ void ha_start_election_async(ha_state_t *ha, cluster_t *c, atomic_fetch_add_explicit(&ha->term, 1, memory_order_acq_rel); uint64_t cur_term = atomic_load_explicit(&ha->term, memory_order_acquire); + /* Record self-vote before sending RequestVote RPCs. This prevents granting + * a vote to a concurrent candidate in the same term (split-brain guard). */ + if (my_name) { + pthread_mutex_lock(&ha->lock); + ha->voted_for_term = cur_term; + strncpy(ha->voted_for_name, my_name, sizeof(ha->voted_for_name) - 1); + ha->voted_for_name[sizeof(ha->voted_for_name) - 1] = '\0'; + pthread_mutex_unlock(&ha->lock); + } + /* Transition to CTRL_ELECTING */ atomic_store_explicit(&ha->role, (int32_t)CTRL_ELECTING, memory_order_release); INFO("HA FSM: → CTRL_ELECTING (async, term=%" PRIu64 ", n_peers=%d)", @@ -2217,29 +2227,37 @@ void ha_on_primary_ctrl_fd_rdhup(ha_state_t *ha, cluster_t *c) if (committed > 0) persist_truncate_wal_above(&c->persist, committed); - /* Clear voted_for so the jitter check below reflects NEW votes cast during - * the RDHUP election, not the stale startup voted_for (different term). */ + /* Clear voted_for: the old primary is dead; any vote from a previous term + * is stale and must not block the new election. */ pthread_mutex_lock(&ha->lock); ha->voted_for_term = 0; memset(ha->voted_for_name, 0, sizeof(ha->voted_for_name)); pthread_mutex_unlock(&ha->lock); - /* Priority-based jitter before RDHUP election to avoid split-brain: - * lower-priority nodes (higher priority index) wait longer, giving the - * highest-priority node time to win before others start their elections. */ + /* Priority-based jitter before RDHUP election. + * The highest-priority (index 0) node starts immediately; lower-priority + * nodes sleep for (priority * election_timeout_ms). During the sleep they + * may receive a RequestVote from the priority-0 node — if so, they grant + * it (setting voted_for) and then skip their own election below. */ if (ha->my_priority > 0) { - uint32_t step_ms = ha->config.election_priority_step_ms - ? ha->config.election_priority_step_ms - : HA_DEFAULT_ELECTION_PRIORITY_STEP_MS; + uint32_t step_ms = ha->config.election_timeout_ms + ? ha->config.election_timeout_ms + : HA_DEFAULT_ELECTION_TIMEOUT_MS; long delay_ns = (long)ha->my_priority * (long)step_ms * 1000000L; struct timespec jitter = { .tv_sec = delay_ns / 1000000000L, .tv_nsec = delay_ns % 1000000000L, }; nanosleep(&jitter, NULL); - /* If already voted during jitter (from the winner's vote request), skip */ + /* Skip our election if we already voted for ANOTHER candidate in the + * current term during the jitter — meaning the priority-0 node sent + * us a RequestVote, we granted it, and should not compete. */ pthread_mutex_lock(&ha->lock); - int already_voted = (ha->voted_for_name[0] != '\0'); + uint64_t cur_term = atomic_load_explicit(&ha->term, memory_order_acquire); + const char *myname_ptr = c && c->myconf ? c->myconf->name : ""; + int already_voted = (ha->voted_for_name[0] != '\0' && + ha->voted_for_term == cur_term && + strcmp(ha->voted_for_name, myname_ptr) != 0); pthread_mutex_unlock(&ha->lock); if (already_voted) { INFO("HA: RDHUP — already voted during jitter, skipping own election"); @@ -2300,7 +2318,9 @@ static void *hb_timeout_thread(void *arg) ha_state_t *ha = t->ha; while (!atomic_load(&t->disarmed)) { - /* Compute deadline: election_timeout_ms + rand_jitter */ + /* Compute deadline: heartbeat_timeout_ms + rand_jitter. + * Must use heartbeat_timeout_ms (not election_timeout_ms) so standbys + * don't falsely trigger failover while the primary is in COMMITTING. */ uint32_t jitter_ms = 0; if (ha->config.election_jitter_max_ms > 0) { struct timespec now; @@ -2308,7 +2328,7 @@ static void *hb_timeout_thread(void *arg) unsigned int seed = (unsigned int)now.tv_nsec; jitter_ms = (uint32_t)(rand_r(&seed) % ha->config.election_jitter_max_ms); } - uint32_t total_ms = ha->config.election_timeout_ms + jitter_ms; + uint32_t total_ms = ha->config.heartbeat_timeout_ms + jitter_ms; struct timespec deadline; clock_gettime(CLOCK_REALTIME, &deadline); diff --git a/src/controller/testsuite/test_ha_failover.c b/src/controller/testsuite/test_ha_failover.c index 249e22d0..9c6d606f 100644 --- a/src/controller/testsuite/test_ha_failover.c +++ b/src/controller/testsuite/test_ha_failover.c @@ -80,6 +80,17 @@ static ha_config_t g_cfg = { .election_priority_step_ms = 500, }; +/* Config with short heartbeat_timeout_ms for watchdog-fires tests */ +static ha_config_t g_fast_cfg = { + .heartbeat_interval_ms = 2000, + .heartbeat_timeout_ms = 2000, /* minimum valid: == heartbeat_interval_ms */ + .election_timeout_ms = 500, + .probe_connect_timeout_ms = 50, + .max_promotion_retries = 3, + .election_jitter_max_ms = 0, + .election_priority_step_ms = 500, +}; + /* ── Test 1: RDHUP closes fd and sets primary_ctrl_fd = -1 ─────────────── */ static void test_rdhup_closes_and_nulls_fd(void) @@ -172,18 +183,19 @@ static void test_heartbeat_received_resets_timeout(void) static void test_heartbeat_timeout_fires_failover(void) { ha_state_t ha; - TEST_ASSERT_EQUAL_INT(0, ha_init(&ha, &g_cfg, 1)); + /* Use fast config: heartbeat_timeout_ms=2000ms so watchdog fires in ~2s */ + TEST_ASSERT_EQUAL_INT(0, ha_init(&ha, &g_fast_cfg, 1)); ha_do_demote(&ha); int sv[2]; TEST_ASSERT_EQUAL_INT(0, socketpair(AF_UNIX, SOCK_STREAM, 0, sv)); ha.primary_ctrl_fd = sv[0]; - /* Arm with 200ms timeout, NO heartbeat received */ + /* Arm watchdog — no heartbeat will arrive */ ha_arm_heartbeat_timeout_vtask(&ha, NULL); - /* Wait 800ms — well past the 500ms election_timeout_ms */ - struct timespec ts = { .tv_sec = 0, .tv_nsec = 800 * 1000000L }; + /* Wait 2500ms — past the 2000ms heartbeat_timeout_ms */ + struct timespec ts = { .tv_sec = 2, .tv_nsec = 500 * 1000000L }; nanosleep(&ts, NULL); /* Failover must have fired: role = CTRL_ELECTING, fd = -1 */ diff --git a/src/include/request.h b/src/include/request.h index a1c11dd8..2eed8fa2 100644 --- a/src/include/request.h +++ b/src/include/request.h @@ -104,10 +104,19 @@ extern int xce_replconfig_reply_ssl_decode(SSL *ssl, size_t *out_len); /* - * xce_upgrade_config_req_extract: extract bytes from a decoded - * OP_upgradeConfig request. Returns -ENOTSUP — see request.c - * for the architectural finding. Use xce_upgrade_config_req_extract_raw - * with the raw packet buffer instead. + * xce_free_request: free an xce_request_t and all owned sub-fields, + * including the raw_sha256/raw_bytes fields on upgrade_config_req_t + * that the generated free_Request() does not know about. + * Use this everywhere instead of calling free_Request() directly. + */ +extern void xce_free_request(xce_request_t *req); + +/* + * xce_upgrade_config_req_extract: extract sha256 + bytes from a decoded + * OP_upgradeConfig request. xce_decode() stores malloc'd copies in + * req->body.upgradeconfigrequestmsg->raw_sha256 / raw_bytes before + * capn_free() runs. Returns shallow pointers (owned by req); caller + * must NOT free them. Returns -EINVAL if fields are missing. */ extern int xce_upgrade_config_req_extract(const xce_request_t *req, uint8_t **out_bytes, diff --git a/src/include/typemap.h b/src/include/typemap.h index dfe1744f..184d9ca5 100644 --- a/src/include/typemap.h +++ b/src/include/typemap.h @@ -348,8 +348,22 @@ struct repl_config_push_reply_s { }; struct upgrade_config_req_s { - xce_data_t *sha256; /* 32-byte SHA-256 of proposed config */ - xce_data_t *bytes; /* serialised ClusterConfig blob */ + /* Legacy decoded fields — populated by generated decode_UpgradeConfigRequest() + * via decode_Data_ptr. These are structurally broken for List(UInt8) wire + * fields (decode_Data_ptr reads a discriminant and misinterprets the bytes), + * but must remain for the generated code to compile. Do NOT use these for + * actual byte extraction. */ + xce_data_t *sha256; /* broken: always a bogus xce_jobs_data_t */ + xce_data_t *bytes; /* broken: always a bogus xce_jobs_data_t */ + + /* Correct raw bytes extracted by xce_decode() before capn_free() runs, + * using UpgradeConfigRequest_get_sha256/bytes + capn_getv8. + * Both are malloc'd and owned here; freed by free_upgrade_config_req_raw() + * called from free_Request() after free_UpgradeConfigRequest(). + * raw_sha256 is exactly 32 bytes when non-NULL. */ + uint8_t *raw_sha256; /* 32-byte SHA-256, malloc'd, or NULL */ + uint8_t *raw_bytes; /* serialised ClusterConfig blob, malloc'd, or NULL */ + size_t raw_bytes_len; /* length of raw_bytes */ }; struct upgrade_config_reply_s { diff --git a/src/lib/request.c b/src/lib/request.c index ba2014b0..a2d57f40 100644 --- a/src/lib/request.c +++ b/src/lib/request.c @@ -69,6 +69,56 @@ int xce_decode(io_packet_t* pkt, xce_request_t* req) decode_Request(req, &rq); + /* OPCode_upgradeConfig carries sha256 and bytes as List(UInt8) on the wire. + * decode_UpgradeConfigRequest() intentionally leaves raw_sha256/raw_bytes as + * NULL (the project's Data union cannot carry raw bytes via decode_Data_ptr). + * Extract them here using UpgradeConfigRequest_get_* + capn_getv8 BEFORE + * capn_free() destroys the capnp segment. */ + if (rq.opcode == OPCode_upgradeConfig + && req->body_which == Request_body_upgradeconfigrequestmsg + && req->body.upgradeconfigrequestmsg != NULL) { + + upgrade_config_req_t *ucr = req->body.upgradeconfigrequestmsg; + UpgradeConfigRequest_ptr u_ptr = rq.body.upgradeconfigrequestmsg; + + /* sha256 */ + { + Data_ptr dp = UpgradeConfigRequest_get_sha256(u_ptr); + capn_list8 sha_list; + sha_list.p = dp.p; + if (sha_list.p.type != CAPN_NULL) { + int slen = capn_len(sha_list); + if (slen > 0) { + uint8_t *sbuf = (uint8_t *)malloc((size_t)slen); + if (sbuf && capn_getv8(sha_list, 0, sbuf, slen) == slen) { + ucr->raw_sha256 = sbuf; + } else { + free(sbuf); + } + } + } + } + + /* bytes */ + { + Data_ptr dp = UpgradeConfigRequest_get_bytes(u_ptr); + capn_list8 bytes_list; + bytes_list.p = dp.p; + if (bytes_list.p.type != CAPN_NULL) { + int blen = capn_len(bytes_list); + if (blen > 0) { + uint8_t *bbuf = (uint8_t *)malloc((size_t)blen); + if (bbuf && capn_getv8(bytes_list, 0, bbuf, blen) == blen) { + ucr->raw_bytes = bbuf; + ucr->raw_bytes_len = (size_t)blen; + } else { + free(bbuf); + } + } + } + } + } + TRACE("%s request op %d seq %ld from %s", __func__, req->opcode, req->seq, REQUEST_FROM(req)); capn_free(&c); @@ -85,6 +135,35 @@ void init_reply_by_req(xce_reply_t* reply, xce_request_t* req) reply->opcode = req->opcode; } +/* + * xce_free_request: free an xce_request_t and ALL owned sub-fields. + * + * The generated free_Request() handles the capnp-decoded fields but does + * not know about raw_sha256/raw_bytes on upgrade_config_req_t (those are + * extra fields populated by xce_decode() before capn_free()). This wrapper + * frees those extra fields first, then calls free_Request(). + * + * Use this everywhere instead of calling free_Request() directly. + */ +void xce_free_request(xce_request_t *req) +{ + if (req == NULL) + return; + + /* Free raw_* fields that the generated free_UpgradeConfigRequest skips */ + if (req->body_which == Request_body_upgradeconfigrequestmsg + && req->body.upgradeconfigrequestmsg != NULL) { + upgrade_config_req_t *ucr = req->body.upgradeconfigrequestmsg; + free(ucr->raw_sha256); + ucr->raw_sha256 = NULL; + free(ucr->raw_bytes); + ucr->raw_bytes = NULL; + ucr->raw_bytes_len = 0; + } + + free_Request(req); +} + int xce_request(endpoint_t* ep, xce_request_t* req) { int rc; @@ -567,31 +646,17 @@ cleanup: } /* - * xce_upgrade_config_req_extract: attempt to extract sha256 + bytes - * from a decoded OP_upgradeConfig request. + * xce_upgrade_config_req_extract: extract sha256 + bytes from a decoded + * OP_upgradeConfig request. * - * ARCHITECTURAL FINDING (Task A research): - * -------------------------------------------------------------------- - * The decoded form is unusable for byte extraction. When iomgr receives - * an OP_upgradeConfig packet, xce_decode() calls decode_Request → - * decode_UpgradeConfigRequest → decode_Data_ptr. decode_Data_ptr - * allocates a fresh xce_data_t and walks decode_Data, which treats the - * pointer as a discriminated Data union (reading a 16-bit discriminator - * at offset 0). But the wire actually carries a List(UInt8) with - * datasz=0, so capn_read16 returns 0 (Data_data_schedjobs) and a - * zero-filled xce_jobs_data_t is allocated. No raw bytes survive. + * xce_decode() now extracts the raw List(UInt8) bytes from the capnp + * segment before capn_free(), storing malloc'd copies in + * req->body.upgradeconfigrequestmsg->raw_sha256 and ->raw_bytes. + * This function returns shallow copies (pointers into the struct); the + * caller must NOT free them — they are owned by the xce_request_t and + * released by free_Request(). * - * Worse: xce_decode() then calls capn_free() and free_packet(), so the - * original capnp segment containing the List(UInt8) bytes is gone by - * the time any handler runs. The xce_data_t* fields on the decoded - * xce_request_t are non-NULL but contain only the bogus jobs_data_t. - * - * This function therefore always returns -ENOTSUP. Task D must use - * xce_upgrade_config_req_extract_raw() below and arrange for the raw - * io_packet_t bytes to reach handle_upgrade_config() before xce_decode - * destroys them — e.g. by adding a pre-decode peek for OP_upgradeConfig - * in the message-handler dispatch, or by stashing the raw buffer on - * xce_request_t for select opcodes. + * Returns 0 on success, -EINVAL on bad args or missing fields. */ int xce_upgrade_config_req_extract(const xce_request_t *req, uint8_t **out_bytes, size_t *out_len, @@ -604,7 +669,22 @@ int xce_upgrade_config_req_extract(const xce_request_t *req, *out_bytes = NULL; *out_len = 0; *out_sha256 = NULL; - return -ENOTSUP; + + if (req->body_which != Request_body_upgradeconfigrequestmsg + || req->body.upgradeconfigrequestmsg == NULL) { + return -EINVAL; + } + + const upgrade_config_req_t *ucr = req->body.upgradeconfigrequestmsg; + if (ucr->raw_sha256 == NULL || ucr->raw_bytes == NULL + || ucr->raw_bytes_len == 0) { + return -EINVAL; + } + + *out_sha256 = ucr->raw_sha256; + *out_bytes = ucr->raw_bytes; + *out_len = ucr->raw_bytes_len; + return 0; } /*