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

fix(election-iomgr-refactor): config replication and RDHUP failover chain

Browse source 
- cluster.c: fix upgradeConfig handler (standbys apply, not primaries)
- cluster.c: fix raw_pkt_copy opcode peek (read scalar before capn_free)
- config_repl.h: enable CONF_ENABLE_UPGRADE_CONFIG=1
- config_repl.c: use client_ssl_ctx for outgoing config fetch connections
- config_repl.c: after config fetch, run cluster_init_phase2 + find_primary
- join.c: push config to newly-joined standbys on OPCode_join
- myapp.c: schedule fetch_config_from_primary after skipping own election
- ha.c: arm heartbeat timeout watchdog after find_and_connect_to_primary
This commit is contained in:
Rongsong Shen 2026-05-31 02:08:48 +00:00
parent 2f5a738b36
commit fd93997f78
6 changed files with 70 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

18
src/controller/cluster.c
View file

@ -159,9 +159,10 @@ static int myreq_handler(vtask_t* task)
xce_request_t request;
xce_reply_t reply;
/* Peek at opcode to decide whether to save raw bytes before xce_decode()
* frees the capnp segment. OPCode_upgradeConfig requires the raw packet
* for xce_upgrade_config_req_extract_raw(). */
/* 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;
{
@ -173,9 +174,12 @@ static int myreq_handler(vtask_t* task)
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_rq.opcode == OPCode_upgradeConfig) {
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);
@ -314,8 +318,10 @@ static int myreq_handler(vtask_t* task)
break;
case OPCode_upgradeConfig:
#if CONF_ENABLE_UPGRADE_CONFIG
if (!ha_is_primary(&c->ha)) {
/* Standby cannot handle upgrade — send error reply */
/* OPCode_upgradeConfig is sent BY the primary TO standbys to push config.
* Standbys must apply it; only a primary sending this to itself is invalid. */
if (ha_is_primary(&c->ha)) {
/* Primary should not receive its own config push — ignore */
xce_reply_t ureply;
init_reply_by_req(&ureply, &request);
ureply.status = -EAGAIN;

22
src/controller/config_repl.c
View file

@ -809,9 +809,13 @@ static int do_one_config_fetch(cluster_t *c, const char *host, uint32_t port,
SSL *ssl = NULL;
int sock = -1;
/* 1. Connect + SSL handshake */
/* 1. Connect + SSL handshake — use client_ssl_ctx (TLS_client_method) for
* outgoing connections; ssl_ctx is the server context (TLS_server_method). */
SSL_CTX *out_ctx = c->ha.client_ssl_ctx
? (SSL_CTX *)c->ha.client_ssl_ctx
: (SSL_CTX *)c->ha.ssl_ctx;
int rc = connect_ssl_raw(
(SSL_CTX *)c->ha.ssl_ctx, host, port,
out_ctx, host, port,
(int)c->ha.config.probe_connect_timeout_ms,
&c->ha, &ssl, &sock);
if (rc != 0)
@ -940,6 +944,20 @@ int fetch_config_from_primary(cluster_t *c)
INFO("HA: config received from primary"
" (tag=%016" PRIx64 ")",
c->myconf->config_version_tag);
/* Config loaded — complete standby bootstrap:
* 1. cluster_init_phase2 to set up data structures with config
* 2. ha_find_and_connect_to_primary to establish the persistent
* control channel needed for RDHUP failover detection */
#ifndef XCE_TESTING
{
extern int cluster_init_phase2(cluster_t *c);
int p2rc = cluster_init_phase2(c);
if (p2rc == 0) {
ha_find_and_connect_to_primary(&c->ha, &c->ha.config,
&c->myconf->peers_cfg, c);
}
}
#endif
return 0;
}

4
src/controller/config_repl.h
View file

@ -16,9 +16,9 @@
/* Sentinel name for the synthetic lost-and-found queue and host */
#define XCE_LOST_AND_FOUND_NAME "lost_and_found"
/* Compile-time gate for OP_upgradeConfig (Phase 2, default disabled) */
/* Compile-time gate for OP_upgradeConfig (enabled: standbys apply pushed config) */
#ifndef CONF_ENABLE_UPGRADE_CONFIG
#define CONF_ENABLE_UPGRADE_CONFIG 0
#define CONF_ENABLE_UPGRADE_CONFIG 1
#endif
/* Push timeout for standby fan-out (ms) */

14
src/controller/ha.c
View file

@ -1203,6 +1203,17 @@ apply_role:
INFO("HA FSM: → CTRL_STANDBY (term=%" PRIu64 ", first_boot=%d)",
atomic_load_explicit(&ha->term, memory_order_acquire),
is_first_boot);
/* Trigger async config fetch from primary so standby can load
* cluster_config, establish persistent control channel, and
* participate in RDHUP failover detection. */
#ifndef XCE_TESTING
if (c != NULL) {
extern int fetch_config_from_primary(cluster_t *c);
delay_task_ms(&c->delaymgr, 100, 1,
(delay_task_entry_fn_t)fetch_config_from_primary, c);
}
#endif
}
}
@ -2123,6 +2134,7 @@ int ha_find_and_connect_to_primary(ha_state_t *ha,
ha->voted_for_name, r);
ha->primary_ctrl_fd = fd;
register_fd_with_iomgr(c, fd);
ha_arm_heartbeat_timeout_vtask(ha, c);
return fd;
}
if (fd == -ETIMEDOUT || fd == -EAGAIN) {
@ -2148,6 +2160,7 @@ int ha_find_and_connect_to_primary(ha_state_t *ha,
INFO("HA: Tier 2 connected to '%s'", p2->name);
ha->primary_ctrl_fd = fd;
register_fd_with_iomgr(c, fd);
ha_arm_heartbeat_timeout_vtask(ha, c);
return fd;
}
}
@ -2164,6 +2177,7 @@ int ha_find_and_connect_to_primary(ha_state_t *ha,
INFO("HA: Tier 3 connected to '%s' after backoff", p3->name);
ha->primary_ctrl_fd = fd;
register_fd_with_iomgr(c, fd);
ha_arm_heartbeat_timeout_vtask(ha, c);
return fd;
}
}

6
src/controller/join.c
View file

@ -5,6 +5,7 @@
#include <unistd.h>
#include "cluster.h"
#include "config_repl.h"
#include "ha.h"
#include "xce/cluster_def.h"
#include "xce/errnox.h"
@ -163,6 +164,11 @@ static int handle_controller_join(cluster_t* c, endpoint_t* ep,
INFO("controller %s joined — %d standby(s) tracked",
request->from ? request->from->name : "?",
c->ha.standby_endpoints->count);
/* Immediately push current config to the newly-joined standby so it
* doesn't have to wait for the next scheduled push. */
if (ha_is_primary(&c->ha))
push_config_to_standbys(c, NULL, 0, NULL, 0);
}
return 0;

16
src/controller/myapp.c
View file

@ -252,6 +252,22 @@ int myapp_run(myconf_t *myconf)
skip_election:;
/* If we skipped starting our own election (already voted for the winner),
* set role to CTRL_STANDBY and trigger config fetch so we can join the cluster. */
{
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&c.ha.role,
memory_order_acquire);
if (role == CTRL_STANDBY || role == CTRL_INIT) {
atomic_store_explicit(&c.ha.role, (int32_t)CTRL_STANDBY,
memory_order_release);
extern int fetch_config_from_primary(cluster_t *c);
/* Use delaymgr one-shot so cluster_run is entered before the
* blocking fetch starts (fetch runs in delaymgr worker thread). */
delay_task_ms(&c.delaymgr, 200, 1,
(delay_task_entry_fn_t)fetch_config_from_primary, &c);
}
}
/* ── Step 9: cluster_run ─────────────────────────────────────────────── */
/* Req 9.4: myapp_run exits into cluster_run within 10ms of startup.