wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity

finalize configuration replication

Browse source
This commit is contained in:
Rongsong Shen 2026-06-09 16:40:13 +08:00
parent e1517bbf28
commit fa66d1ba9c
30 changed files with 8867 additions and 1560 deletions
Download patch file Download diff file Expand all files Collapse all files

2
src/controller/CLAUDE.md
View file

@ -160,7 +160,7 @@ VERBOSE=-vvv /tmp/xce-deploy/scripts/start.sh ctrl-3 &
# Primary elected and config pushed
grep -i "CTRL_PRIMARY\|Config committed" /tmp/xce-deploy/logs/ctrl-1.log | grep -v DEBUG3
# Config files present on standbys
find /tmp/xce-deploy/ctrl-2/conf /tmp/xce-deploy/ctrl-3/conf -type f
find /tmp/xce-deploy/controller/ctrl-2/conf /tmp/xce-deploy/controller/ctrl-3/conf -type f
```
**Clean restart** (keeps ctrl-1 config + all certs, wipes data/standby-conf/logs):

20
src/controller/ha.c
View file

@ -3544,6 +3544,9 @@ static void on_standby_connected(endpoint_t *ep)
pthread_mutex_unlock(&ha->lock);
}
/* Reset failure counter so the next disconnect restarts logging from attempt 1. */
ha->peer_connect_fail_count[sctx->peer_idx] = 0;
/* Record last_push_time before any send so the first heartbeat tick after
* connect does not immediately send a redundant NOP (Req 3.3, task 2.2). */
{
@ -3639,10 +3642,18 @@ static void ha_connect_to_peer(cluster_t *c, int peer_idx)
endpoint_set_on_connected(ep, on_standby_connected);
endpoint_set_err_handler(ep, standby_ep_err_handler_primary);
int n = ++ha->peer_connect_fail_count[peer_idx];
int should_log = (n == 1) || (n % HA_CONNECT_LOG_THROTTLE == 0);
if (should_log)
INFO("HA: peer[%d] (%s) not connected — initiating outbound connect (attempt %d)",
peer_idx, peer->hostname, n);
int rc = endpoint_connect(ep, peer->hostname, peer->port);
if (rc < 0) {
WARN("HA: connect to peer[%d] (%s:%u) failed: %d",
peer_idx, peer->hostname, peer->port, rc);
if (should_log)
WARN("HA: connect to peer[%d] (%s:%u) failed: %d",
peer_idx, peer->hostname, peer->port, rc);
/* endpoint_fini calls FREEUP(ep->data) which frees sctx — do not
* double-free sctx here. */
ep->data = NULL;
@ -3754,11 +3765,8 @@ static int primary_heartbeat_fn(void *arg)
pthread_mutex_unlock(&ha->lock);
}
if (!found) {
INFO("HA: peer[%d] (%s) not connected — initiating outbound connect",
i, pcfg->candidates[i].name ? pcfg->candidates[i].name : "?");
if (!found)
ha_connect_to_peer(c, i);
}
}
}

3
src/controller/ha.h
View file

@ -123,6 +123,8 @@ typedef struct {
#define HA_DEFAULT_HEARTBEAT_TIMEOUT_MS 10000
#define HA_DEFAULT_ELECTION_TIMEOUT_MS 5000
#define HA_DEFAULT_PROBE_CONNECT_TIMEOUT_MS 500
/* Log the first connect failure per peer, then every Nth attempt. */
#define HA_CONNECT_LOG_THROTTLE 10
#define HA_DEFAULT_MAX_PROMOTION_RETRIES 5
#define HA_DEFAULT_ELECTION_JITTER_MAX_MS 200
#define HA_DEFAULT_ELECTION_PRIORITY_STEP_MS 500
@ -189,6 +191,7 @@ typedef struct {
uint64_t standby_reported_tag[MAX_HA_CANDIDATES]; /* Last config_tag reported by each standby */
uint64_t standby_reported_lsn[MAX_HA_CANDIDATES]; /* Last applied_lsn reported by each standby */
bool standby_converged[MAX_HA_CANDIDATES]; /* Whether standby[i] has converged with primary */
int peer_connect_fail_count[MAX_HA_CANDIDATES]; /* throttle log spam on repeated connect failure */
} ha_state_t;
/* ── Public API ────────────────────────────────────────────── */

22
src/controller/main.c
View file

@ -28,6 +28,7 @@ const char* argp_program_bug_address = BUG_EMAIL;
#define OPT_CONF_KEEP_N 257
#define OPT_CONF_STALE_TIMEOUT 258
#define OPT_VCPUS 259
#define OPT_CERTDIR 260
struct arguments {
char* myname;
@ -38,6 +39,7 @@ struct arguments {
int conf_keep_n;
long conf_stale_timeout;
int vcpus;
char* certdir;
};
static struct argp_option options[] = {
@ -54,6 +56,8 @@ static struct argp_option options[] = {
"Seconds before saved config is flagged stale (0=disabled, default: 86400)" },
{ "vcpus", OPT_VCPUS, "N", 0,
"Number of virtual CPUs for the task manager (must be >= 1)" },
{ "certdir", OPT_CERTDIR, "DIR", 0,
"Directory containing xce.crt and xce.key (default: <workdir>/certs)" },
{ 0 }
};
@ -93,6 +97,9 @@ static error_t parse_opt(int key, char* arg, struct argp_state* state)
if (validate_vcpus(arguments->vcpus) != 0)
argp_error(state, "--vcpus must be >= 1 (got: %s)", arg);
break;
case OPT_CERTDIR:
arguments->certdir = arg;
break;
default:
return ARGP_ERR_UNKNOWN;
}
@ -147,8 +154,15 @@ int main(int argc, char* argv[])
INFO("My name is %s", arguments.myname);
snprintf(key_file, XCE_MAXPATH_LEN, "%s/certs/xce.key", arguments.workdir);
snprintf(cert_file, XCE_MAXPATH_LEN, "%s/certs/xce.crt", arguments.workdir);
const char *certdir = arguments.certdir;
char certdir_buf[XCE_MAXPATH_LEN + 1];
if (!certdir) {
snprintf(certdir_buf, sizeof(certdir_buf),
"%s/certs", arguments.workdir);
certdir = certdir_buf;
}
snprintf(key_file, XCE_MAXPATH_LEN, "%s/xce.key", certdir);
snprintf(cert_file, XCE_MAXPATH_LEN, "%s/xce.crt", certdir);
myconf.cluster_config = NULL;
myconf.conffile = NULL; /* set by load_primary_config() or fetch_config_from_primary() */
@ -180,10 +194,10 @@ int main(int argc, char* argv[])
return 1;
}
/* CA cert derived from workdir — same infrastructure as certfile/keyfile */
/* CA cert from certdir (same certificate as certfile) */
snprintf(myconf.peers_cfg.ca_cert_path,
sizeof(myconf.peers_cfg.ca_cert_path),
"%s/certs/xce.crt", arguments.workdir);
"%s/xce.crt", certdir);
strncpy(myconf.peers_cfg.identity, arguments.myname,
sizeof(myconf.peers_cfg.identity) - 1);
INFO("HA mode: %d controller candidates, identity=%s",