forked from lei.chen/myxce
Rongsong Shen
c97c7632e9
Root cause (found via ha-convergence-recovery task 8 live 3-node E2E): a controller with no local config (config_version_tag==0) could become PRIMARY and push an empty config, breaking replication. Two upstream election-safety gaps, both outside ha-convergence-recovery: 1. RDHUP re-election bypassed the no-config 'voter only' guard. The startup path (ha_start_election_async) refuses to start an election when my_config_tag==0 && n_peers>1, but the RDHUP failover path (ha_on_primary_ctrl_fd_rdhup) called raft_on_leader_lost unconditionally. Fix: mirror the guard before elect_ctx alloc / hook / raft_on_leader_lost; a config-less node stays CTRL_STANDBY (voter only) and grants the configured peer instead of starting (and winning) its own election. 2. Voter freshness gate compared the candidate only against the voter's LOCAL config, ignoring the cluster's known committed state. Fix: ha_vote_self_freshness floors the comparison at max(local, learned-committed config_tag / committed_lsn), so a voter denies any candidate whose config/cluster state is outdated vs what the cluster has committed. Strictly additive (never denies the freshest node — Leader Completeness preserves liveness). TDD: test_rdhup_noconfig_node_does_not_elect + configured-node contrast; test_ha_vote_gate learned-floor deny + liveness + backward-compat. Full controller suite green under ASAN. commit.c and ha-convergence-recovery files untouched.
1570 lines
50 KiB
C
1570 lines
50 KiB
C
#include "config.h"
|
||
|
||
#include <errno.h>
|
||
#include <inttypes.h>
|
||
#include <rtti.h>
|
||
#include <stdatomic.h>
|
||
#include <string.h>
|
||
#include <unistd.h>
|
||
|
||
#include "cluster.h"
|
||
#include "config_repl.h"
|
||
#include "controller.capnp.h"
|
||
#include "dataq.h"
|
||
#include "htab.h"
|
||
#include "kvmgr.h"
|
||
#include "mylock.h"
|
||
#include "request.h"
|
||
#include "vtaskmgr.h"
|
||
#include "xce/cluster_def.h"
|
||
#include "zarray.h"
|
||
|
||
#include "xce/types.h"
|
||
|
||
#include "jobj.h"
|
||
#include "resobj.h"
|
||
#include "utils.h"
|
||
#include "worker.h"
|
||
#include "xce/errnox.h"
|
||
#include "xstring.h"
|
||
#include "helper.h"
|
||
|
||
/* startup_validation.c — forward declaration */
|
||
int validate_heartbeat_config(uint32_t heartbeat_timeout_ms,
|
||
uint32_t heartbeat_interval_ms,
|
||
char *err_buf, size_t err_len);
|
||
|
||
static xce_peerinfo_t from_controller
|
||
= { .name = "controller", .key = "keepSecret" };
|
||
|
||
static ep_private_t* ep_get_private()
|
||
{
|
||
ep_private_t* priv;
|
||
|
||
priv = (ep_private_t*)calloc(1, sizeof(*priv));
|
||
return priv;
|
||
}
|
||
|
||
static int handle_client_leave(cluster_t* c) { return 0; }
|
||
|
||
static void handle_endpoint_error(endpoint_t* ep)
|
||
{
|
||
cluster_t* c = (cluster_t*)ep->owner;
|
||
ep_private_t* priv = NULL;
|
||
int role = -1;
|
||
|
||
if (c == NULL) {
|
||
ERROR("endpoint %p has no owner cluster", ep);
|
||
|
||
return;
|
||
}
|
||
|
||
priv = (ep_private_t*)ep->data;
|
||
|
||
TRACE("endpoint %p priv data %p", ep, priv);
|
||
|
||
if (priv != NULL) {
|
||
role = priv->role;
|
||
switch (role) {
|
||
case HostRole_worker:
|
||
if (priv->u.idx >= 0) {
|
||
handle_worker_leave(c, priv->u.idx);
|
||
} else {
|
||
ERROR("incorrect worker idx %d", priv->u.idx);
|
||
}
|
||
break;
|
||
case HostRole_scheduler:
|
||
handle_scheduler_leave(c);
|
||
break;
|
||
case HostRole_controller:
|
||
if (c->ha.standby_endpoints) {
|
||
pthread_mutex_lock(&c->ha.lock);
|
||
for (int idx = 0;
|
||
idx < c->ha.standby_endpoints->count; idx++) {
|
||
endpoint_t *se = (endpoint_t *)zarray_get(
|
||
c->ha.standby_endpoints, idx);
|
||
if (se == ep) {
|
||
zarray_del(c->ha.standby_endpoints, idx);
|
||
break;
|
||
}
|
||
}
|
||
pthread_mutex_unlock(&c->ha.lock);
|
||
}
|
||
break;
|
||
case HostRole_client:
|
||
handle_client_leave(c);
|
||
break;
|
||
case HostRole_apiserver:
|
||
handle_apiserver_leave(c, ep);
|
||
break;
|
||
default:
|
||
ERROR("unknown role %d for endpoint %p", priv->role, ep);
|
||
break;
|
||
}
|
||
}
|
||
|
||
if (ep->idx >= 0) {
|
||
/* zarray_del will finalize ep */
|
||
zarray_del(&(c->endpoints), ep->idx);
|
||
return;
|
||
}
|
||
|
||
endpoint_fini(ep);
|
||
FREEUP(ep);
|
||
|
||
return;
|
||
}
|
||
|
||
static int do_notify(cluster_t* c, endpoint_t* ep, xce_request_t* req)
|
||
{
|
||
xce_notify_data_t* data = req->body.notifymsg;
|
||
|
||
switch (data->event) {
|
||
case NotifyData_data_jobPidInfo:
|
||
on_job_start(c, data->data.pidinfo);
|
||
break;
|
||
case NotifyData_data_jobFailInfo:
|
||
on_job_fail(c, data->data.failinfo);
|
||
break;
|
||
case NotifyData_data_jobFinishInfo:
|
||
on_job_finish(c, data->data.finishinfo);
|
||
break;
|
||
case NotifyData_data_stoppedJob:
|
||
on_job_stop(c, data->data.stopped_job);
|
||
break;
|
||
case NotifyData_data_resumedJob:
|
||
on_job_resume(c, data->data.resumed_job);
|
||
break;
|
||
case NotifyData_data_ckptinfo:
|
||
on_job_ckpt(c, ep, data->data.ckptinfo);
|
||
break;
|
||
case NotifyData_data_postDoneJob:
|
||
on_job_postdone(c, ep, data->data.pdone_job);
|
||
break;
|
||
case NotifyData_data_taskPidInfo:
|
||
on_task_start(c, data->data.taskinfo);
|
||
break;
|
||
case NotifyData_data_taskFailInfo:
|
||
on_task_fail(c, data->data.taskfailinfo);
|
||
break;
|
||
case NotifyData_data_taskFinish:
|
||
on_task_finish(c, data->data.taskfinish);
|
||
break;
|
||
default:
|
||
break;
|
||
}
|
||
return 0;
|
||
}
|
||
/* handle_wal_catchup_entry: dispatch handler for OPCode_walCatchupEntry.
|
||
* Delegates to ha_apply_wal_catchup_entry (ha.c); on error logs and returns 0
|
||
* (skip, no crash). Standby remains in CTRL_STANDBY throughout (Req 3.4).
|
||
* Task 3.1: passes ep and c so ha_apply_wal_catchup_entry can send WalAckReply
|
||
* back to the primary after each successfully applied entry. */
|
||
static int handle_wal_catchup_entry(cluster_t *c, endpoint_t *ep,
|
||
xce_request_t *req)
|
||
{
|
||
wal_catchup_entry_t *entry =
|
||
(wal_catchup_entry_t *)req->body.walcatchupentrymsg;
|
||
|
||
/* body may be NULL if the sender used Request_body_nomsg; skip gracefully */
|
||
if (!entry)
|
||
return 0;
|
||
|
||
/* Req 1.3 (ha-heartbeat-liveness): any msg from primary resets standby
|
||
* liveness and steps down from CTRL_ELECTING. */
|
||
{
|
||
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(
|
||
&c->ha.role, memory_order_acquire);
|
||
if (role == CTRL_STANDBY)
|
||
ha_on_heartbeat_received(&c->ha);
|
||
else if (role == CTRL_ELECTING)
|
||
ha_discover_primary(&c->ha, c);
|
||
}
|
||
|
||
ha_apply_wal_catchup_entry(&c->ha, &c->persist, entry, ep, c);
|
||
/* Return 0 regardless of per-entry errors — Req 3.3: skip, don't crash */
|
||
return 0;
|
||
}
|
||
|
||
|
||
/* handle_snapshot_transfer: dispatch handler for OPCode_snapshotTransfer.
|
||
* Delegates to ha_handle_snapshot_chunk (ha.c).
|
||
* On mid-transfer reconnect the caller (iomgr reconnect path) must delete
|
||
* snap.part before the next transfer starts — this handler does not track
|
||
* cross-chunk state itself (Req 4.4).
|
||
* No reply is sent — primary learns of completion via next heartbeat reply. */
|
||
static int handle_snapshot_transfer(cluster_t *c, xce_request_t *req)
|
||
{
|
||
snapshot_transfer_t *chunk =
|
||
(snapshot_transfer_t *)req->body.snapshottransfermsg;
|
||
|
||
/* body may be NULL if sender used Request_body_nomsg; skip gracefully */
|
||
if (!chunk)
|
||
return 0;
|
||
|
||
/* Req 1.3 (ha-heartbeat-liveness): any msg from primary resets standby
|
||
* liveness and steps down from CTRL_ELECTING. */
|
||
{
|
||
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(
|
||
&c->ha.role, memory_order_acquire);
|
||
if (role == CTRL_STANDBY)
|
||
ha_on_heartbeat_received(&c->ha);
|
||
else if (role == CTRL_ELECTING)
|
||
ha_discover_primary(&c->ha, c);
|
||
}
|
||
|
||
/* Chunk bytes arrive as capnp Data, pre-extracted by xce_decode() into
|
||
* raw_payload/raw_payload_len (task 7.2). */
|
||
int rc = ha_handle_snapshot_chunk(&c->ha, &c->persist,
|
||
c->persist.snap_path,
|
||
chunk,
|
||
chunk->raw_payload,
|
||
chunk->raw_payload_len);
|
||
if (rc != 0)
|
||
ERROR("handle_snapshot_transfer: chunk %u rc=%d",
|
||
chunk->chunk_index, rc);
|
||
/* Return 0 — errors are logged; primary retries on next ctrlhello (Req 4.4) */
|
||
return 0;
|
||
}
|
||
|
||
static int myreq_handler(vtask_t* task)
|
||
{
|
||
vtask_data_t* data = (vtask_data_t*)task->data;
|
||
endpoint_t* ep = data->u.iodata.ep;
|
||
cluster_t* c = (cluster_t*)ep->owner;
|
||
int stop = 0;
|
||
int rc = 0;
|
||
xce_request_t request;
|
||
xce_reply_t reply;
|
||
|
||
/* 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,
|
||
xce_request_op_name(&request));
|
||
|
||
init_reply_by_req(&reply, &request);
|
||
|
||
/* Dispatch gate: reject ops not allowed in current FSM state (req 1.3, 1.4) */
|
||
rc = ha_dispatch_gate(&c->ha, request.opcode);
|
||
if (rc != 0) {
|
||
reply.status = rc;
|
||
reply.data_which = RequestReply_data_nodata;
|
||
xce_reply(ep, c->myhost, &reply);
|
||
xce_free_request(&request);
|
||
return rc;
|
||
}
|
||
|
||
switch (request.opcode) {
|
||
case OPCode_shutdown:
|
||
reply.status = 0;
|
||
reply.data_which = RequestReply_data_nodata;
|
||
xce_reply(ep, c->myhost, &reply);
|
||
stop = 1;
|
||
cbuf_put(&(c->wbuf), &stop);
|
||
break;
|
||
case OPCode_hello:
|
||
if (request.body_which == Request_body_ctrlhellomsg) {
|
||
/* Extended OP_ctrlHello with CtrlHelloRequest body — state sync (Req 6.5–6.7) */
|
||
rc = handle_ctrlhello(c, ep, &request);
|
||
/* Req 4.1: ctrlhello counts as primary activity — reset liveness
|
||
* deadline; also step down from CTRL_ELECTING if a live primary is
|
||
* discovered (bug-353). */
|
||
{
|
||
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(
|
||
&c->ha.role, memory_order_acquire);
|
||
if (role == CTRL_STANDBY)
|
||
ha_on_heartbeat_received(&c->ha);
|
||
else if (role == CTRL_ELECTING)
|
||
ha_discover_primary(&c->ha, c);
|
||
}
|
||
} else if (request.body_which == Request_body_replymsg
|
||
&& request.body.replymsg
|
||
&& request.body.replymsg->data_which
|
||
== RequestReply_data_ctrlhelloreplymsg) {
|
||
/* Standby receives CtrlHelloReply from primary — update last known
|
||
* committed LSN and config tag for use in RDHUP failover path. */
|
||
ctrl_hello_reply_t *chrep = request.body.replymsg->data.ctrlhelloreplymsg;
|
||
if (chrep)
|
||
cluster_apply_ctrlhello_reply(c, chrep->committed_lsn,
|
||
chrep->config_tag);
|
||
} else {
|
||
/* Legacy plain OP_hello (nop path) */
|
||
reply.status = 0;
|
||
reply.data_which = RequestReply_data_nodata;
|
||
xce_reply(ep, c->myhost, &reply);
|
||
}
|
||
break;
|
||
case OPCode_nop: {
|
||
/* Task 2.3: standby populates stateReport in every heartbeat reply so
|
||
* the primary can track convergence (Req 1.1, 1.3, 1.4). */
|
||
state_report_t sr;
|
||
memset(&sr, 0, sizeof(sr));
|
||
reply.status = 0;
|
||
ha_nop_reply_state_report(&c->ha, c->myconf, &sr, &reply);
|
||
xce_reply(ep, c->myhost, &reply);
|
||
/* Req 1.3: any NOP from primary resets liveness deadline on standby.
|
||
* Also steps down from CTRL_ELECTING if a live primary is discovered. */
|
||
{
|
||
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(
|
||
&c->ha.role, memory_order_acquire);
|
||
if (role == CTRL_STANDBY)
|
||
ha_on_heartbeat_received(&c->ha);
|
||
else if (role == CTRL_ELECTING)
|
||
ha_discover_primary(&c->ha, c);
|
||
}
|
||
break;
|
||
}
|
||
case OPCode_join:
|
||
rc = handle_join(c, ep, &request);
|
||
break;
|
||
case OPCode_jobdispatch:
|
||
rc = handle_jobdispatch(c, ep, &request);
|
||
break;
|
||
case OPCode_jobsub:
|
||
rc = handle_jobsubmit(c, ep, &request);
|
||
break;
|
||
case OPCode_get:
|
||
rc = do_get_op(c, ep, &request);
|
||
break;
|
||
case OPCode_notify:
|
||
rc = do_notify(c, ep, &request);
|
||
break;
|
||
case OPCode_clusterconf:
|
||
rc = handle_clusterconf(c, ep, &request);
|
||
break;
|
||
case OPCode_update:
|
||
rc = do_update_op(c, ep, &request);
|
||
break;
|
||
case OPCode_jobcontrol:
|
||
rc = do_jobcontrol_op(c, ep, &request);
|
||
break;
|
||
case OPCode_queuecontrol:
|
||
rc = do_queuecontrol_op(c, ep, &request);
|
||
break;
|
||
case OPCode_hostcontrol:
|
||
rc = do_hostcontrol_op(c, ep, &request);
|
||
break;
|
||
case OPCode_runtasks:
|
||
rc = do_runtasks(c, ep, &request);
|
||
break;
|
||
case OPCode_setpreason:
|
||
rc = do_set_preason(c, ep, &request);
|
||
break;
|
||
case OPCode_clrpreason:
|
||
rc = do_clr_preason(c, ep, &request);
|
||
break;
|
||
case OPCode_requestVote: {
|
||
/* Task 6.3: delegate vote-request handling to raft library.
|
||
*
|
||
* The freshness gate (config_tag / log_lsn comparison) is applied here
|
||
* BEFORE calling raft_on_vote_request because raft_cb_vote_eligible
|
||
* (on_vote_eligible) intentionally skips the freshness check — it
|
||
* relies on the caller (this dispatcher) to enforce it first.
|
||
*
|
||
* raft_on_vote_request handles:
|
||
* - stale-term deny (term < current_term)
|
||
* - term upgrade (term > current_term → clear voted_for)
|
||
* - role gate: RAFT_LEADER denies (via on_vote_eligible)
|
||
* - one-vote-per-term invariant (voted_for != -1 && voted_for != from_idx)
|
||
* - on_persist before granting (durability guarantee)
|
||
*
|
||
* reply_term is set by raft_on_vote_request to the current term
|
||
* (after any term upgrade), so vr_rep.term reflects the latest term. */
|
||
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 local_tag = c->myconf->config_version_tag;
|
||
uint64_t local_lsn = (uint64_t)atomic_load_explicit(
|
||
&c->ha.applied_lsn, memory_order_acquire);
|
||
|
||
/* Election-safety fix #2: the freshness floor is the max of our local
|
||
* state and our learned-committed cluster state, so we deny candidates
|
||
* that are outdated against what the cluster has already committed
|
||
* (not merely against our local config file / applied LSN). */
|
||
uint64_t my_tag = 0, my_lsn = 0;
|
||
ha_vote_self_freshness(&c->ha, local_tag, local_lsn, &my_tag, &my_lsn);
|
||
|
||
/* Freshness gate: deny if candidate is not at least as fresh as the
|
||
* effective floor.
|
||
* (Req 2.4: config_tag must be ≥ ours; if equal, log_lsn must be ≥ ours.) */
|
||
int freshness_ok = !(cand_tag < my_tag ||
|
||
(cand_tag == my_tag && cand_lsn < my_lsn));
|
||
|
||
/* Map candidate name to integer index for raft_on_vote_request. */
|
||
int from_idx = -1;
|
||
const char *cand_id = vrq ? vrq->candidate_id : "";
|
||
if (cand_id && cand_id[0] != '\0') {
|
||
const xce_cluster_config_t *pcfg = &c->myconf->peers_cfg;
|
||
for (int i = 0; i < pcfg->candidates_count; i++) {
|
||
if (strcmp(pcfg->candidates[i].name, cand_id) == 0) {
|
||
from_idx = i;
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
|
||
uint64_t reply_term = 0;
|
||
int granted = 0;
|
||
if (freshness_ok && from_idx >= 0) {
|
||
int r = raft_on_vote_request(&c->ha.raft, from_idx, cand_term,
|
||
&reply_term);
|
||
granted = (r == RAFT_GRANT) ? 1 : 0;
|
||
} else {
|
||
/* Freshness deny or unknown candidate: read current term for reply */
|
||
reply_term = atomic_load_explicit(&c->ha.term, memory_order_acquire);
|
||
}
|
||
|
||
request_vote_reply_t vr_rep;
|
||
memset(&vr_rep, 0, sizeof(vr_rep));
|
||
vr_rep.term = reply_term;
|
||
vr_rep.vote_granted = granted;
|
||
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);
|
||
/* Inbound vote connections are for voting only. After the reply is
|
||
* sent the connection is left open until the remote side closes it.
|
||
* The primary will connect outbound to the newcomer from the heartbeat
|
||
* task if the peer is missing from standby_endpoints. */
|
||
|
||
rc = 0;
|
||
break;
|
||
}
|
||
case OPCode_replConfig:
|
||
rc = handle_config_request(c, ep, &request);
|
||
break;
|
||
case OPCode_replConfigPush:
|
||
/* Primary pushes config to this standby — save and reload */
|
||
rc = handle_repl_config_push(c, ep, &request);
|
||
break;
|
||
case OPCode_walCatchupEntry:
|
||
/* Task 2.1: standby applies a single WAL catch-up entry from the
|
||
* primary. applied_lsn advances on success; errors are logged and
|
||
* skipped (Req 3.1, 3.3, 3.4, 3.5). No reply is sent — primary
|
||
* learns of progress via the next heartbeat reply's stateReport. */
|
||
rc = handle_wal_catchup_entry(c, ep, &request);
|
||
break;
|
||
case OPCode_snapshotTransfer:
|
||
/* Task 3.5c: standby receives one snapshot chunk from the primary.
|
||
* Writes chunk payload to snap.part; on isLast renames and loads.
|
||
* No reply — primary learns of completion via next heartbeat reply.
|
||
* Req 4.2, 4.3, 4.4. */
|
||
rc = handle_snapshot_transfer(c, &request);
|
||
break;
|
||
case OPCode_upgradeConfig:
|
||
#if CONF_ENABLE_UPGRADE_CONFIG
|
||
/* 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;
|
||
ureply.data_which = RequestReply_data_nodata;
|
||
xce_reply(ep, c->myhost, &ureply);
|
||
rc = -EAGAIN;
|
||
break;
|
||
}
|
||
rc = handle_upgrade_config(c, ep, &request);
|
||
break;
|
||
#else
|
||
{
|
||
xce_reply_t ureply;
|
||
init_reply_by_req(&ureply, &request);
|
||
ureply.status = -ENOSYS;
|
||
ureply.data_which = RequestReply_data_nodata;
|
||
xce_reply(ep, c->myhost, &ureply);
|
||
rc = -ENOSYS;
|
||
}
|
||
break;
|
||
#endif
|
||
case OPCode_reply:
|
||
/* Config-push ACK arriving on an inbound standby endpoint (the
|
||
* join.c-registered standby_endpoints channel — these eps reply on
|
||
* the same connection the primary pushed on, so the ACK lands here
|
||
* instead of peer_ctrl_msg_handler). Feed it to the replicator so
|
||
* the configAck barrier can observe it (ha-commit-barrier-rollout
|
||
* task 5.2 remediation r1; same ingestion as
|
||
* repl_ack_decode_and_dispatch in ha.c — completes the unlanded
|
||
* half of ha-config-bootstrap task 8.1). */
|
||
if (request.body_which == Request_body_replymsg
|
||
&& request.body.replymsg
|
||
&& request.body.replymsg->data_which
|
||
== RequestReply_data_replconfigpushreplymsg
|
||
&& request.body.replymsg->data.replconfigpushreplymsg) {
|
||
repl_config_push_reply_t *cack =
|
||
request.body.replymsg->data.replconfigpushreplymsg;
|
||
if (cack->accepted) {
|
||
repl_on_config_ack(&c->persist.repl, ep, cack->version_tag);
|
||
TRACE("myreq_handler: config ACK from inbound standby"
|
||
" tag=%016" PRIx64, cack->version_tag);
|
||
} else {
|
||
WARN("myreq_handler: config push rejected by inbound standby"
|
||
" (tag=%016" PRIx64 ") — ignoring", cack->version_tag);
|
||
}
|
||
rc = 0;
|
||
break;
|
||
}
|
||
/* Any other reply body keeps the legacy unknown-opcode handling. */
|
||
/* fall through */
|
||
default:
|
||
rc = -E_XCE_BAD_OPCODE;
|
||
ERROR("Unknown opcode: %d", request.opcode);
|
||
reply.status = rc;
|
||
reply.data_which = RequestReply_data_nodata;
|
||
xce_reply(ep, c->myhost, &reply);
|
||
|
||
break;
|
||
}
|
||
|
||
xce_free_request(&request);
|
||
|
||
return rc;
|
||
}
|
||
|
||
static int vtask_entry(vtask_t* task)
|
||
{
|
||
int rc;
|
||
|
||
rc = myreq_handler(task);
|
||
|
||
return rc;
|
||
}
|
||
|
||
static int handle_msg(endpoint_t* ep)
|
||
{
|
||
cluster_t* c = NULL;
|
||
|
||
if (ep == NULL) {
|
||
return -EINVAL;
|
||
}
|
||
|
||
c = (cluster_t*)ep->owner;
|
||
if (c == NULL) {
|
||
return -EINVAL;
|
||
}
|
||
|
||
return vtask_handle_msg(&(c->vtaskmgr), (vtask_func_t)vtask_entry,
|
||
(void*)ep);
|
||
}
|
||
|
||
static int handle_connection(struct epoll_event* p)
|
||
{
|
||
endpoint_t* ep = (endpoint_t*)p->data.ptr;
|
||
endpoint_t* ep2;
|
||
cluster_t* c = (cluster_t*)ep->owner;
|
||
iomgr_t* iomgr = &(c->iomgr);
|
||
int idx;
|
||
ep_private_t* priv = NULL;
|
||
|
||
ep2 = endpoint_get();
|
||
if (ep2 == NULL) {
|
||
PERROR("endpoint_get");
|
||
|
||
return -ENOMEM;
|
||
}
|
||
|
||
endpoint_accept(ep, ep2);
|
||
if (ep2->fd < 0) {
|
||
PERROR("endpoint_accept");
|
||
return -1;
|
||
}
|
||
|
||
TRACE("server sock %d accepted %d\n", ep->fd, ep2->fd);
|
||
|
||
priv = ep_get_private();
|
||
if (priv == NULL) {
|
||
PERROR("ep_get_private");
|
||
|
||
endpoint_fini(ep2);
|
||
FREEUP(ep2);
|
||
|
||
return -ENOMEM;
|
||
}
|
||
|
||
/* Assume client as default */
|
||
priv->role = HostRole_client;
|
||
priv->u.idx = -1;
|
||
|
||
idx = zarray_store(&(c->endpoints), ep2);
|
||
if (idx < 0) {
|
||
SET_ERROR(-idx);
|
||
|
||
PERROR("zarray_store");
|
||
|
||
FREEUP(priv);
|
||
endpoint_fini(ep2);
|
||
FREEUP(ep2);
|
||
|
||
return idx;
|
||
}
|
||
|
||
ep2->idx = idx;
|
||
ep2->data = priv;
|
||
ep2->iomgr = iomgr;
|
||
|
||
endpoint_set_owner(ep2, c);
|
||
endpoint_set_peer(ep2, c->myhost);
|
||
endpoint_set_msg_handler(ep2, (msg_handler_t)handle_msg);
|
||
endpoint_set_err_handler(ep2, (err_handler_t)handle_endpoint_error);
|
||
|
||
iomgr_monitor(ep2);
|
||
return 0;
|
||
}
|
||
|
||
static int controller_init(controller_t* c, cluster_t* cluster)
|
||
{
|
||
c->ep = NULL;
|
||
|
||
return 0;
|
||
}
|
||
|
||
static void endpoint_killer(void* p)
|
||
{
|
||
endpoint_t* ep = (endpoint_t*)p;
|
||
cluster_t* c = (cluster_t*)ep->owner;
|
||
|
||
if (ep->idx >= 0) {
|
||
zarray_clear(&(c->endpoints), ep->idx);
|
||
ep->idx = -1;
|
||
}
|
||
endpoint_fini(ep);
|
||
FREEUP(ep);
|
||
|
||
return;
|
||
}
|
||
|
||
int cluster_build_config(cluster_t* c)
|
||
{
|
||
char valbuf[128];
|
||
int i;
|
||
|
||
snprintf(valbuf, sizeof(valbuf) - 1, "%d",
|
||
c->myconf->cluster_config->heartbeat_interval);
|
||
add_config(c, "heartbeat_interval", valbuf);
|
||
snprintf(valbuf, sizeof(valbuf) - 1, "%d",
|
||
c->myconf->cluster_config->inactive_heartbeat_count);
|
||
add_config(c, "inactive_heartbeat_count", valbuf);
|
||
snprintf(valbuf, sizeof(valbuf) - 1, "%d",
|
||
c->myconf->cluster_config->ncandidates);
|
||
add_config(c, "ncandidates", valbuf);
|
||
|
||
for (i = 0; i < c->myconf->cluster_config->ncandidates; i++) {
|
||
char keybuf[128];
|
||
|
||
sprintf(keybuf, "candidate_%d_name", i);
|
||
sprintf(valbuf, "%s", c->myconf->cluster_config->candidates[i]->name);
|
||
|
||
add_config(c, keybuf, valbuf);
|
||
|
||
sprintf(keybuf, "candidate_%d_host", i);
|
||
sprintf(valbuf, "%s", c->myconf->cluster_config->candidates[i]->host);
|
||
|
||
add_config(c, keybuf, valbuf);
|
||
sprintf(keybuf, "candidate_%d_port", i);
|
||
sprintf(valbuf, "%d", c->myconf->cluster_config->candidates[i]->port);
|
||
|
||
add_config(c, keybuf, valbuf);
|
||
}
|
||
|
||
return 0;
|
||
}
|
||
|
||
int cluster_build_resources(cluster_t* c)
|
||
{
|
||
int i;
|
||
|
||
if (c->myconf == NULL) {
|
||
return -EINVAL;
|
||
}
|
||
|
||
for (i = 0; i < c->myconf->cluster_config->nres; i++) {
|
||
xce_resource_t* res;
|
||
|
||
res = makeResource(c->resmgr, c->myconf->cluster_config->resources[i]);
|
||
if (res == NULL) {
|
||
return -errno;
|
||
}
|
||
}
|
||
|
||
return 0;
|
||
}
|
||
|
||
int cluster_load_jobs(cluster_t* c)
|
||
{
|
||
load_jobs(c);
|
||
|
||
return 0;
|
||
}
|
||
|
||
int cluster_build_queues(cluster_t* c)
|
||
{
|
||
build_queues(c);
|
||
return 0;
|
||
}
|
||
|
||
int cluster_build_groups(cluster_t* c)
|
||
{
|
||
/* TODO: implement group building */
|
||
return 0;
|
||
}
|
||
|
||
static int xce_config_init(int type, int situation, void* data, va_list ap)
|
||
{
|
||
int rc = 0;
|
||
cluster_t* c = va_arg(ap, cluster_t*);
|
||
|
||
if (situation == HOOK_BEFORE_EVENT) {
|
||
c->myconf->cluster_config = get_default_conf(c);
|
||
if (c->myconf->cluster_config == NULL) {
|
||
return -EINVAL;
|
||
}
|
||
return 0;
|
||
}
|
||
|
||
xce_use(c);
|
||
|
||
rc = cluster_build_config(c);
|
||
if (rc < 0) {
|
||
return rc;
|
||
}
|
||
rc = cluster_build_resources(c);
|
||
if (rc < 0) {
|
||
return rc;
|
||
}
|
||
|
||
rc = cluster_build_queues(c);
|
||
if (rc < 0) {
|
||
return rc;
|
||
}
|
||
rc = cluster_build_groups(c);
|
||
if (rc < 0) {
|
||
return rc;
|
||
}
|
||
|
||
rc = cluster_load_jobs(c);
|
||
if (rc < 0) {
|
||
return rc;
|
||
}
|
||
|
||
tlog_info("xce controller is running");
|
||
|
||
return rc;
|
||
}
|
||
|
||
static int cluster_hook_fun(int type, int situation, int detail, void* data,
|
||
va_list ap)
|
||
{
|
||
if (detail == HOOK_DETAIL_CLUSTER_INIT) {
|
||
return xce_config_init(type, situation, data, ap);
|
||
}
|
||
|
||
return 0;
|
||
}
|
||
|
||
static int xce_install_hooks(cluster_t* c)
|
||
{
|
||
xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_CLUSTER,
|
||
(hook_fn_t)cluster_hook_fun, NULL);
|
||
return 0;
|
||
}
|
||
|
||
static int run_delay_tasks(delay_task_entry_fn_t entry, void* arg)
|
||
{
|
||
return entry(arg);
|
||
}
|
||
|
||
static void res_obj_killer(xce_resource_t* r)
|
||
{
|
||
if (r == NULL) {
|
||
return;
|
||
}
|
||
}
|
||
|
||
static void job_obj_killer(void* arg)
|
||
{
|
||
if (arg == NULL) {
|
||
return;
|
||
}
|
||
}
|
||
|
||
static void worker_obj_killer(void* arg)
|
||
{
|
||
if (arg == NULL) {
|
||
return;
|
||
}
|
||
}
|
||
|
||
static void queue_obj_killer(void* arg)
|
||
{
|
||
if (arg == NULL) {
|
||
return;
|
||
}
|
||
}
|
||
|
||
static void resval_obj_killer(void* arg)
|
||
{
|
||
if (arg == NULL) {
|
||
return;
|
||
}
|
||
}
|
||
|
||
static void config_obj_killer(void* arg)
|
||
{
|
||
if (arg == NULL) {
|
||
return;
|
||
}
|
||
}
|
||
|
||
/*
|
||
* init_controller_tunables: set performance-tuning defaults on c->tunables.
|
||
*
|
||
* Called early in cluster_init_phase1 so the values are available throughout
|
||
* the startup sequence (including pre-election and phase2).
|
||
*
|
||
* TODO: wire these to cluster_config fields once the config generation tool
|
||
* supports jobmgr/workermgr/qmgr/resmgr/xstrings (@13–@17).
|
||
*/
|
||
/*
|
||
* cluster_init_managers: allocate and initialize the manager subsystems.
|
||
*
|
||
* Called from cluster_init_phase1 so these are ready before election —
|
||
* handle_connection() arrives during the pre-election phase and may call
|
||
* zarray_store on c->endpoints (via endpoint tracking) which requires the
|
||
* managers to exist. cluster_init_phase2 skips re-init if already done.
|
||
*/
|
||
static int cluster_init_managers(cluster_t *c)
|
||
{
|
||
kvmgr_params_t config_mgr_params
|
||
= { .key_type = KVMGR_KEY_STRING,
|
||
.nslots = 128,
|
||
.delta = 128,
|
||
.datakiller = (destroy_fn_t)config_obj_killer };
|
||
kvmgr_params_t resmgr_params
|
||
= { .key_type = KVMGR_KEY_STRING,
|
||
.nslots = c->tunables.resmgr_nhash,
|
||
.delta = c->tunables.resmgr_delta,
|
||
.datakiller = (destroy_fn_t)res_obj_killer };
|
||
kvmgr_params_t jobmgr_params
|
||
= { .key_type = KVMGR_KEY_INT,
|
||
.nslots = c->tunables.jobmgr_nhash,
|
||
.delta = c->tunables.jobmgr_delta,
|
||
.datakiller = (destroy_fn_t)job_obj_killer };
|
||
kvmgr_params_t workermgr_params
|
||
= { .key_type = KVMGR_KEY_STRING,
|
||
.nslots = c->tunables.workermgr_nhash,
|
||
.delta = c->tunables.workermgr_delta,
|
||
.datakiller = (destroy_fn_t)worker_obj_killer };
|
||
kvmgr_params_t qmgr_params
|
||
= { .key_type = KVMGR_KEY_STRING,
|
||
.nslots = c->tunables.qmgr_nhash,
|
||
.delta = c->tunables.qmgr_delta,
|
||
.datakiller = (destroy_fn_t)queue_obj_killer };
|
||
|
||
c->configmgr = makeInstance(&xkvmgr_type, &config_mgr_params);
|
||
c->jobmgr = makeInstance(&xkvmgr_type, &jobmgr_params);
|
||
c->resmgr = makeInstance(&xkvmgr_type, &resmgr_params);
|
||
zarray_init(&(c->sharedinst), c->tunables.resmgr_inst,
|
||
(zarray_item_killer_t)resval_obj_killer);
|
||
c->workermgr = makeInstance(&xkvmgr_type, &workermgr_params);
|
||
c->qmgr = makeInstance(&xkvmgr_type, &qmgr_params);
|
||
|
||
xstring_table_init(c->tunables.xstrings_nhash);
|
||
job_obj_init(5381, 5381);
|
||
|
||
zarray_init(&(c->endpoints), 16, endpoint_killer);
|
||
return 0;
|
||
}
|
||
|
||
static void init_controller_tunables(cluster_t *c)
|
||
{
|
||
/* Jobs can be numerous; workers/queues/resources are typically smaller.
|
||
* TODO: replace with config-driven values once the config generation tool
|
||
* supports cluster_config fields @13–@17 (jobmgr/workermgr/qmgr/resmgr/xstrings). */
|
||
c->tunables.jobmgr_nhash = 512;
|
||
c->tunables.jobmgr_delta = 128;
|
||
c->tunables.workermgr_nhash = 128;
|
||
c->tunables.workermgr_delta = 32;
|
||
c->tunables.qmgr_nhash = 128;
|
||
c->tunables.qmgr_delta = 32;
|
||
c->tunables.resmgr_nhash = 128;
|
||
c->tunables.resmgr_delta = 32;
|
||
c->tunables.resmgr_inst = 64;
|
||
c->tunables.xstrings_nhash = 512;
|
||
}
|
||
|
||
int cluster_init_phase1(cluster_t* c)
|
||
{
|
||
int rc;
|
||
|
||
if (c == NULL)
|
||
return -EINVAL;
|
||
|
||
/* Populate myhost_info so replies identify this node by its real name
|
||
* (e.g. ctrl-1) rather than the generic "controller" placeholder. */
|
||
c->myhost_info.name = (c->myconf && c->myconf->name)
|
||
? c->myconf->name : from_controller.name;
|
||
c->myhost_info.key = from_controller.key;
|
||
c->myhost = &c->myhost_info;
|
||
c->ncontrollers = 1;
|
||
c->myidx = 0;
|
||
|
||
xce_hook_init(&(c->hooksys), XCE_CONTROLLER_HOOK_MAX);
|
||
xce_install_hooks(c);
|
||
|
||
/* HA config and priority — derived from --peers list before config loads */
|
||
ha_config_t ha_cfg = {
|
||
.heartbeat_interval_ms = HA_DEFAULT_HEARTBEAT_INTERVAL_MS,
|
||
.heartbeat_timeout_ms = HA_DEFAULT_HEARTBEAT_TIMEOUT_MS,
|
||
.election_timeout_ms = HA_DEFAULT_ELECTION_TIMEOUT_MS,
|
||
.probe_connect_timeout_ms = HA_DEFAULT_PROBE_CONNECT_TIMEOUT_MS,
|
||
.max_promotion_retries = HA_DEFAULT_MAX_PROMOTION_RETRIES,
|
||
.election_jitter_max_ms = HA_DEFAULT_ELECTION_JITTER_MAX_MS,
|
||
.election_priority_step_ms = HA_DEFAULT_ELECTION_PRIORITY_STEP_MS,
|
||
};
|
||
|
||
int my_priority = 0;
|
||
const xce_cluster_config_t* pcfg = &c->myconf->peers_cfg;
|
||
if (pcfg->candidates_count > 0) {
|
||
my_priority = pcfg->candidates_count - 1; /* default: last = lowest */
|
||
for (int i = 0; i < pcfg->candidates_count; i++) {
|
||
const char* key = pcfg->candidates[i].name[0]
|
||
? pcfg->candidates[i].name
|
||
: pcfg->candidates[i].hostname;
|
||
if (strncmp(key, c->myconf->name, XCE_MAXNAME_LEN - 1) == 0) {
|
||
my_priority = i;
|
||
break;
|
||
}
|
||
}
|
||
INFO("HA peers: %d candidates, my_priority=%d (name=%s)",
|
||
pcfg->candidates_count, my_priority, c->myconf->name);
|
||
}
|
||
|
||
/* Req 2.4: reject misconfigured heartbeat timing before ha_init() */
|
||
{
|
||
char hb_err[256];
|
||
if (validate_heartbeat_config(ha_cfg.heartbeat_timeout_ms,
|
||
ha_cfg.heartbeat_interval_ms,
|
||
hb_err, sizeof(hb_err)) != 0) {
|
||
ERROR("startup validation failed: %s", hb_err);
|
||
fprintf(stderr, "xce_controller: %s\n", hb_err);
|
||
return -EINVAL;
|
||
}
|
||
}
|
||
|
||
rc = ha_init(&c->ha, &ha_cfg, my_priority);
|
||
if (rc != 0) {
|
||
ERROR("ha_init failed: %d", rc);
|
||
return rc;
|
||
}
|
||
|
||
/* Restore persisted election term + voted_for from ha-term.bin (Req 9.2).
|
||
* File lives under {workdir}/{myname}/data/ alongside WAL and snapshot. */
|
||
if (c->myconf->workdir && c->myconf->name) {
|
||
char term_dir[1024];
|
||
snprintf(term_dir, sizeof(term_dir), "%s/%s/data",
|
||
c->myconf->workdir, c->myconf->name);
|
||
rc = ha_restore_term(&c->ha, term_dir);
|
||
if (rc == 0) {
|
||
INFO("HA: restored term=%lu from %s/ha-term.bin",
|
||
(unsigned long)atomic_load(&c->ha.term), term_dir);
|
||
} else if (rc != -ENOENT) {
|
||
WARN("HA: ha_restore_term failed: %d", rc);
|
||
}
|
||
}
|
||
|
||
/* ha_install_sigusr1() removed — failover_requested replaced by ha_arm_heartbeat_timeout_vtask */
|
||
|
||
/* Initialize standby endpoint tracking */
|
||
c->ha.standby_endpoints = (zarray_t *)calloc(1, sizeof(zarray_t));
|
||
if (c->ha.standby_endpoints) {
|
||
zarray_init(c->ha.standby_endpoints, 4, NULL);
|
||
}
|
||
|
||
/* cbuf used by cluster_run() to signal restart/stop */
|
||
cbuf_init(&(c->wbuf), 1, sizeof(int));
|
||
|
||
/* Sentinel: -1 means iomgr not yet initialised.
|
||
* cluster_init_phase2() uses this to skip re-init when myapp_run()
|
||
* has already called iomgr_init() before election. */
|
||
c->iomgr.epoll_fd = -1;
|
||
|
||
/* Sentinel: fd=-1 means the listener socket is not yet open.
|
||
* cluster_open_listener() uses this for its idempotency guard. */
|
||
c->ep.fd = -1;
|
||
|
||
init_controller_tunables(c);
|
||
cluster_init_managers(c);
|
||
|
||
return 0;
|
||
}
|
||
|
||
int cluster_open_listener(cluster_t* c)
|
||
{
|
||
int rc;
|
||
|
||
if (c == NULL)
|
||
return -EINVAL;
|
||
|
||
/* Idempotent: if the listener socket is already open, do nothing. */
|
||
if (c->ep.fd >= 0)
|
||
return 0;
|
||
|
||
/* Determine listen port: prefer per-candidate port from --peers list,
|
||
* fall back to cluster_config->port if config is already loaded. */
|
||
uint32_t my_port = 0;
|
||
{
|
||
const xce_cluster_config_t* pcfg = &c->myconf->peers_cfg;
|
||
for (int j = 0; j < pcfg->candidates_count; j++) {
|
||
const char* key = pcfg->candidates[j].name[0]
|
||
? pcfg->candidates[j].name
|
||
: pcfg->candidates[j].hostname;
|
||
if (strncmp(key, c->myconf->name, XCE_MAXNAME_LEN - 1) == 0) {
|
||
my_port = pcfg->candidates[j].port;
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
if (my_port == 0) {
|
||
if (c->myconf->cluster_config != NULL) {
|
||
my_port = (uint32_t)c->myconf->cluster_config->port;
|
||
} else {
|
||
ERROR("cluster_open_listener: no port in --peers and cluster_config "
|
||
"not yet loaded; specify port in --peers list");
|
||
return -EINVAL;
|
||
}
|
||
}
|
||
|
||
rc = endpoint_init(&(c->ep), AF_INET, SOCK_STREAM, (int)my_port, 5);
|
||
if (rc < 0)
|
||
return rc;
|
||
|
||
endpoint_set_peer(&(c->ep), c->myhost);
|
||
return 0;
|
||
}
|
||
|
||
int cluster_init_phase2(cluster_t* c)
|
||
{
|
||
int i, rc;
|
||
|
||
if (c == NULL)
|
||
return -EINVAL;
|
||
|
||
if (c->myconf->cluster_config == NULL) {
|
||
ERROR("cluster_init_phase2: cluster_config not loaded");
|
||
return -EINVAL;
|
||
}
|
||
|
||
SSL_CTX* ssl_ctx = (SSL_CTX*)c->ha.ssl_ctx;
|
||
if (ssl_ctx == NULL) {
|
||
ERROR("cluster_init_phase2: ssl_ctx not set (myapp_run must set "
|
||
"c->ha.ssl_ctx before calling phase2)");
|
||
return -EINVAL;
|
||
}
|
||
|
||
/* HOOK_BEFORE_EVENT for CLUSTER_INIT loads cluster_config from disk via
|
||
* xce_config_init(). In the two-phase path, cluster_config is already
|
||
* populated by load_primary_config() / fetch_config_from_primary() before
|
||
* phase2 is called — so we do NOT re-run the hook here.
|
||
* The legacy cluster_init() wrapper runs it before calling phase2. */
|
||
|
||
mylock_init(&(c->sched.lock));
|
||
c->sched.ep = NULL;
|
||
|
||
/* Managers, xstrings, endpoints, job_obj are initialized early in
|
||
* cluster_init_phase1 via cluster_init_managers(). Skip re-init here. */
|
||
|
||
/* iomgr and vtaskmgr are initialised in cluster_open_listener() /
|
||
* myapp_run() before election. Guard against re-initialisation when
|
||
* cluster_init() (legacy wrapper) calls cluster_init_phase2() directly. */
|
||
if (c->iomgr.epoll_fd < 0) {
|
||
iomgr_init(&(c->iomgr));
|
||
iomgr_set_ssl_ctx(&(c->iomgr), ssl_ctx);
|
||
} else {
|
||
/* ssl_ctx may have been set before phase1; always refresh it here. */
|
||
iomgr_set_ssl_ctx(&(c->iomgr), ssl_ctx);
|
||
}
|
||
|
||
/* Open the listener socket (idempotent — skipped if already opened before
|
||
* election via cluster_open_listener()). */
|
||
rc = cluster_open_listener(c);
|
||
if (rc < 0)
|
||
return rc;
|
||
|
||
for (i = 0; i < c->ncontrollers; i++) {
|
||
controller_init(&(c->controllers[i]), c);
|
||
}
|
||
|
||
zarray_init(&(c->configs), 64, NULL /* FIXME */);
|
||
|
||
/* Req 4.3: use --vcpus CLI value; fall back to built-in default (8).
|
||
* The cluster config 'vcpus' field has been removed (Req 4.4 deprecation).
|
||
* vtaskmgr_init() is guarded by n_vcpus==0: myapp_run() calls it before
|
||
* election; the guard here covers the legacy cluster_init() path where
|
||
* it has not been called yet. */
|
||
if (c->vtaskmgr.n_vcpus == 0) {
|
||
int nvcpus = (c->myconf->vcpus > 0) ? c->myconf->vcpus : 8;
|
||
vtaskmgr_init(&(c->vtaskmgr), nvcpus);
|
||
}
|
||
|
||
/* Guard: myapp_run() calls delaymgr_init + delaymgr_start before election
|
||
* (needed for election timeout watchdog via delay_task_ms).
|
||
* Skip re-init on the two-phase path; run on the legacy cluster_init() path
|
||
* where delaymgr has not been initialized yet (precision_ms == 0). */
|
||
if (c->delaymgr.precision_ms == 0) {
|
||
delaymgr_init(&(c->delaymgr), 200 /* precision_ms */, run_delay_tasks);
|
||
}
|
||
|
||
apiserver_init(c);
|
||
|
||
/* Persistence: derive paths and init WAL/snapshot directories.
|
||
* Recovery is deferred until after cluster_build_resources() runs inside
|
||
* the HOOK_AFTER_EVENT below so that resmgr is fully populated before
|
||
* restore_worker() tries to register per-host resource values.
|
||
*
|
||
* Guard: myapp_run() calls persist_set_workdir + persist_init + persist_start
|
||
* before election (so WAL is ready when cluster_init_phase2 fires from vtask
|
||
* context after election). Skip re-init if wal_path is already set. */
|
||
if (c->persist.wal_path[0] == '\0') {
|
||
const char* workdir = c->myconf->workdir ? c->myconf->workdir
|
||
: DEFAULT_WORKDIR;
|
||
const char* name = c->myconf->name ? c->myconf->name
|
||
: "controller";
|
||
persist_config_t pcfg = persist_config_defaults();
|
||
|
||
persist_set_workdir(&c->persist, workdir, name);
|
||
rc = persist_init(&c->persist, &pcfg, c);
|
||
if (rc != 0) {
|
||
ERROR("persist_init failed: %d", rc);
|
||
return rc;
|
||
}
|
||
}
|
||
|
||
/* ── Commit engine instantiation (ha-commit-engine task 5.3) ──────────
|
||
* Both inputs are final exactly here: n_vcpus was resolved by the
|
||
* vtaskmgr_init guard above (or by myapp_run before election) and
|
||
* ha->n_peers was fixed by ha_init_raft before the election that
|
||
* triggered this phase2. Legacy cluster_init() path (tests /
|
||
* single-node): ha->n_peers is still 0 → standalone self-quorum
|
||
* semantics, matching the old degraded single-node contract.
|
||
*
|
||
* Wire the engine BEFORE the promotion below: standbys register for
|
||
* replication only after the post-promotion ctrlhello, so r->ce is set
|
||
* once before any replication traffic (repl.h set-once contract).
|
||
*
|
||
* Guard c->commit.p: phase2 can run twice on the legacy wrapper path
|
||
* (cluster_init() directly, then the first-boot election callback);
|
||
* the engine must be initialized exactly once. */
|
||
if (c->commit.p == NULL) {
|
||
rc = commit_engine_init(&c->commit, &c->persist, c->ha.n_peers,
|
||
c->vtaskmgr.n_vcpus);
|
||
if (rc != 0) {
|
||
ERROR("commit_engine_init failed: %d (n_peers=%d, n_vcpus=%d)",
|
||
rc, c->ha.n_peers, c->vtaskmgr.n_vcpus);
|
||
return rc;
|
||
}
|
||
repl_set_commit_engine(&c->persist.repl, &c->commit);
|
||
/* Write-admission gate reads registry truth through this
|
||
* back-pointer (ha-commit-engine task 4.2 / Req 3.1-3.3). */
|
||
c->ha.repl = &c->persist.repl;
|
||
INFO("commit engine wired (n_peers=%d, n_vcpus=%d)",
|
||
c->ha.n_peers, c->vtaskmgr.n_vcpus);
|
||
}
|
||
|
||
/* Role set by ha_elect_evaluate_and_signal() inside cluster_run() */
|
||
if (ha_is_primary(&c->ha))
|
||
ha_promote_to_primary(c);
|
||
else
|
||
atomic_store_explicit(&c->ha.role, (int32_t)CTRL_STANDBY,
|
||
memory_order_release);
|
||
|
||
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_CLUSTER, HOOK_AFTER_EVENT,
|
||
HOOK_DETAIL_CLUSTER_INIT, c);
|
||
|
||
/* Ensure lost_and_found synthetic queue/host exist before recovery */
|
||
create_lost_and_found(c);
|
||
|
||
/* Recover WAL/snapshot now that resmgr, jobmgr, and workermgr are
|
||
* fully populated by cluster_build_resources() above. */
|
||
rc = persist_recover(&c->persist);
|
||
if (rc != 0) {
|
||
ERROR("persist_recover failed: %d", rc);
|
||
persist_fini(&c->persist);
|
||
return rc;
|
||
}
|
||
|
||
return 0;
|
||
}
|
||
|
||
int cluster_init(cluster_t* c)
|
||
{
|
||
int rc;
|
||
|
||
/* Legacy wrapper: run both phases with SSL init inline.
|
||
* Used by tests and any caller that provides a fully populated myconf_t
|
||
* (including cluster_config) before calling cluster_init(). */
|
||
if (c == NULL)
|
||
return -EINVAL;
|
||
|
||
rc = cluster_init_phase1(c);
|
||
if (rc != 0)
|
||
return rc;
|
||
|
||
/* Create SSL context for the wrapper path */
|
||
initialize_openssl();
|
||
SSL_CTX* ssl_ctx = create_ssl_context(c->myconf->certfile,
|
||
c->myconf->keyfile);
|
||
if (ssl_ctx == NULL) {
|
||
ERROR("create_ssl_context failed");
|
||
cluster_fini(c);
|
||
return -errno;
|
||
}
|
||
c->ha.ssl_ctx = ssl_ctx;
|
||
|
||
/* For the legacy path, run the before-event hook to populate
|
||
* cluster_config via xce_config_init() if it isn't already set */
|
||
if (c->myconf->cluster_config == NULL) {
|
||
rc = xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_CLUSTER,
|
||
HOOK_BEFORE_EVENT, HOOK_DETAIL_CLUSTER_INIT, c);
|
||
if (rc < 0) {
|
||
cluster_fini(c);
|
||
return rc;
|
||
}
|
||
}
|
||
|
||
rc = cluster_init_phase2(c);
|
||
if (rc != 0) {
|
||
cluster_fini(c);
|
||
return rc;
|
||
}
|
||
return 0;
|
||
}
|
||
|
||
static int make_heartbeat(cluster_t* c, endpoint_t* ep)
|
||
{
|
||
return xce_nop_req(ep);
|
||
}
|
||
|
||
static int check_worker_timeout_(cluster_t* c, worker_t* w)
|
||
{
|
||
worker_t* w1 = NULL;
|
||
endpoint_t* ep;
|
||
time_t now = time(NULL);
|
||
int timeout;
|
||
cluster_config_t* config;
|
||
|
||
if (w == NULL) {
|
||
return 0;
|
||
}
|
||
|
||
if (c == NULL) {
|
||
return -EINVAL;
|
||
}
|
||
|
||
ep = w->ep;
|
||
if (ep == NULL) {
|
||
return 0;
|
||
}
|
||
|
||
/* grab lock on worker */
|
||
|
||
config = c->myconf->cluster_config;
|
||
timeout = config->heartbeat_interval * config->inactive_heartbeat_count;
|
||
|
||
if (now - ep->ts_last_in > timeout) {
|
||
TRACE("%s worker %s becomes UNAVAIL as heartbeat timeout", __func__,
|
||
w->host.name);
|
||
|
||
mark_worker_unavail(c, w);
|
||
|
||
return 0;
|
||
} else {
|
||
if (IS_WORKER_UNAVAIL(w->host.status)) {
|
||
mark_worker_back(c, w);
|
||
}
|
||
}
|
||
|
||
if (now - ep->ts_last_out >= config->heartbeat_interval) {
|
||
if (DATAQ_EMPTY(&(ep->outq))) {
|
||
int rc = 0;
|
||
|
||
TRACE("%s send heartbeat to %s", __func__, w->host.name);
|
||
|
||
rc = make_heartbeat(c, ep);
|
||
|
||
if (rc < 0) {
|
||
ERROR("%s make_heartbeat to %s failed", __func__, w->host.name);
|
||
|
||
return rc;
|
||
}
|
||
}
|
||
}
|
||
|
||
return 0;
|
||
}
|
||
|
||
static int check_workers_timeout(void* args)
|
||
{
|
||
iterator_t it;
|
||
cluster_t* c = (cluster_t*)args;
|
||
int rc = 0;
|
||
xlist_t l;
|
||
int i;
|
||
|
||
xlist_init(&l, 1024);
|
||
|
||
lock(c->workermgr, RTTI_LOCK_READ);
|
||
rc = iter_make(&it, c->workermgr);
|
||
if (rc < 0) {
|
||
unlock(c->workermgr);
|
||
xlist_fini(&l);
|
||
|
||
return rc;
|
||
}
|
||
|
||
for (;;) {
|
||
worker_t* w = NULL;
|
||
|
||
rc = iter_next(&it, (void**)&w);
|
||
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
|
||
break;
|
||
}
|
||
|
||
if (w->ep == NULL) {
|
||
continue;
|
||
}
|
||
|
||
/* check_worker_timeout_(c, w); */
|
||
if (w != NULL) {
|
||
reference(w);
|
||
|
||
xlist_store(&l, w);
|
||
}
|
||
}
|
||
|
||
if (rc == RTTI_ITER_NO_DATA) {
|
||
rc = 0;
|
||
}
|
||
|
||
iter_destroy(&it);
|
||
unlock(c->workermgr);
|
||
|
||
for (i = 0; i < l.count; i++) {
|
||
worker_t* w = xlist_get(&l, i);
|
||
|
||
if (w == NULL) {
|
||
continue;
|
||
}
|
||
|
||
lock(w, RTTI_LOCK_WRITE);
|
||
check_worker_timeout_(c, w);
|
||
unlock(w);
|
||
|
||
dereference(w);
|
||
}
|
||
|
||
xlist_fini(&l);
|
||
|
||
return rc;
|
||
}
|
||
|
||
/*
|
||
* cluster_register_listener: wire c->ep into iomgr's epoll watch.
|
||
*
|
||
* Called before election from myapp_run() so that incoming OP_requestVote
|
||
* SSL connections are accepted and dispatched during the election phase.
|
||
* cluster_start() is idempotent — it calls iomgr_monitor again but
|
||
* iomgr_monitor is safe to call twice (epoll EPOLL_CTL_MOD handles it).
|
||
*/
|
||
int cluster_register_listener(cluster_t *c)
|
||
{
|
||
endpoint_set_owner(&(c->ep), c);
|
||
endpoint_set_conn_handler(&(c->ep), (conn_handler_t)handle_connection);
|
||
c->ep.iomgr = &(c->iomgr);
|
||
return iomgr_monitor(&(c->ep));
|
||
}
|
||
|
||
int cluster_start(cluster_t* c)
|
||
{
|
||
int rc;
|
||
|
||
endpoint_set_owner(&(c->ep), c);
|
||
endpoint_set_conn_handler(&(c->ep), (conn_handler_t)handle_connection);
|
||
c->ep.iomgr = &(c->iomgr);
|
||
|
||
/* Skip iomgr_monitor if cluster_register_listener already registered
|
||
* this fd with epoll (ep->flags is set on successful registration). */
|
||
if (c->ep.flags == 0) {
|
||
rc = iomgr_monitor(&(c->ep));
|
||
if (rc < 0)
|
||
return rc;
|
||
}
|
||
|
||
/* Guard: myapp_run() starts iomgr and vtaskmgr before election so that
|
||
* incoming OP_requestVote RPCs are accepted during the election phase.
|
||
* Only start here if they were not already started (legacy cluster_init path). */
|
||
if (c->iomgr.worker == 0)
|
||
iomgr_start(&(c->iomgr));
|
||
if (c->vtaskmgr.dispatcher == 0)
|
||
vtaskmgr_start(&(c->vtaskmgr));
|
||
delaymgr_start(&(c->delaymgr));
|
||
|
||
/* Start persistence fsync thread and register WAL hooks */
|
||
rc = persist_start(&c->persist);
|
||
if (rc != 0) {
|
||
ERROR("persist_start failed: %d", rc);
|
||
return rc;
|
||
}
|
||
|
||
/* ha_start_watchdog() removed — watchdog pthread replaced by
|
||
* ha_arm_heartbeat_timeout_vtask (delaymgr) */
|
||
|
||
delay_task(&(c->delaymgr), c->myconf->cluster_config->heartbeat_interval,
|
||
-1, (delay_task_entry_fn_t)check_workers_timeout, (void*)c);
|
||
|
||
c->running = 1;
|
||
return 0;
|
||
}
|
||
|
||
int cluster_run(cluster_t* c)
|
||
{
|
||
for (;;) {
|
||
int stop;
|
||
|
||
cbuf_get(&(c->wbuf), &stop);
|
||
if (stop) {
|
||
break;
|
||
}
|
||
}
|
||
|
||
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_CLUSTER, HOOK_BEFORE_EVENT,
|
||
HOOK_DETAIL_CLUSTER_SHUTDOWN, c);
|
||
c->running = 0;
|
||
|
||
delaymgr_stop(&(c->delaymgr));
|
||
|
||
iomgr_unmonitor(&(c->iomgr), &(c->ep));
|
||
iomgr_stop(&(c->iomgr));
|
||
|
||
if (atomic_load(&c->restart_requested)) {
|
||
INFO("Config upgrade restart requested — exiting for service manager restart");
|
||
}
|
||
|
||
return 0;
|
||
}
|
||
|
||
/* ── Task 4.2: ha_send_ctrlhello_to_all_standbys ───────────────────────────
|
||
*
|
||
* Send OP_hello CtrlHelloRequest to every connected standby endpoint.
|
||
* This triggers handle_ctrlhello on the receiving standby, which runs the
|
||
* existing catch-up replication sequence (WAL streaming, config push).
|
||
*
|
||
* Must be called from the post-election path in ha_elect_evaluate_and_signal
|
||
* after ha->role is set to CTRL_PRIMARY (Req 7.2).
|
||
*
|
||
* Thread safety: acquires ha->lock around standby_endpoints iteration.
|
||
* Best-effort: failures per-endpoint are logged but do not abort the loop.
|
||
*/
|
||
void ha_send_ctrlhello_to_all_standbys(cluster_t *c)
|
||
{
|
||
if (!c) return;
|
||
ha_state_t *ha = &c->ha;
|
||
if (!ha->standby_endpoints) return;
|
||
|
||
uint64_t my_lsn = atomic_load_explicit(&c->persist.committed_lsn,
|
||
memory_order_acquire);
|
||
uint64_t my_tag = c->myconf ? c->myconf->config_version_tag : 0;
|
||
char *my_name = c->myconf ? c->myconf->name : "";
|
||
|
||
pthread_mutex_lock(&ha->lock);
|
||
int n = ha->standby_endpoints->count;
|
||
for (int i = 0; i < n; i++) {
|
||
endpoint_t *ep = (endpoint_t *)zarray_get(ha->standby_endpoints, i);
|
||
if (!ep) continue;
|
||
|
||
ctrl_hello_req_t hello = {
|
||
.name = my_name,
|
||
.key = NULL,
|
||
.applied_lsn = my_lsn,
|
||
.config_tag = my_tag,
|
||
};
|
||
|
||
xce_request_t req;
|
||
memset(&req, 0, sizeof(req));
|
||
req.opcode = OPCode_hello;
|
||
req.body_which = Request_body_ctrlhellomsg;
|
||
req.body.ctrlhellomsg = &hello;
|
||
|
||
int rc = xce_request(ep, &req);
|
||
if (rc < 0) {
|
||
WARN("ha_send_ctrlhello_to_all_standbys: xce_request to standby[%d] "
|
||
"failed: %d", i, rc);
|
||
}
|
||
}
|
||
pthread_mutex_unlock(&ha->lock);
|
||
}
|
||
|
||
void cluster_fini(cluster_t* c)
|
||
{
|
||
/* Free heap-allocated standby_endpoints array (created in phase1) */
|
||
if (c->ha.standby_endpoints) {
|
||
zarray_fini(c->ha.standby_endpoints);
|
||
free(c->ha.standby_endpoints);
|
||
c->ha.standby_endpoints = NULL;
|
||
}
|
||
|
||
/* iomgr owns the SSL_CTX lifetime; null ha's copy so ha_fini doesn't double-free */
|
||
c->ha.ssl_ctx = NULL;
|
||
/* Stop HA watchdog thread before tearing down other subsystems — P1-5 */
|
||
ha_fini(&c->ha);
|
||
|
||
/* Stop persistence before tearing down the rest */
|
||
persist_stop(&c->persist);
|
||
persist_fini(&c->persist);
|
||
|
||
delaymgr_fini(&(c->delaymgr));
|
||
vtaskmgr_fini(&(c->vtaskmgr));
|
||
endpoint_fini(&(c->ep));
|
||
|
||
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_CLUSTER, HOOK_AFTER_EVENT,
|
||
HOOK_DETAIL_CLUSTER_SHUTDOWN, c);
|
||
cbuf_fini(&(c->wbuf));
|
||
iomgr_stop(&(c->iomgr)); /* join worker thread before freeing SSL_CTX */
|
||
iomgr_fini(&(c->iomgr));
|
||
|
||
/* Commit engine teardown (ha-commit-engine task 5.3): destroy only after
|
||
* ALL replication traffic has stopped — vtaskmgr_fini joined the
|
||
* job-submit vtasks (commit_submit callers) and iomgr_stop/iomgr_fini
|
||
* joined the I/O thread (commit_on_ack source). r->ce is never NULLed
|
||
* while traffic lives (repl.h set-once contract); guard makes this safe
|
||
* when phase2 never ran (early init failure). */
|
||
if (c->commit.p != NULL)
|
||
commit_engine_fini(&c->commit);
|
||
|
||
destroy(c->jobmgr);
|
||
destroy(c->workermgr);
|
||
destroy(c->qmgr);
|
||
destroy(c->resmgr);
|
||
destroy(c->configmgr);
|
||
mylock_fini(&(c->sched.lock));
|
||
|
||
zarray_fini(&(c->endpoints));
|
||
zarray_fini(&(c->configs));
|
||
|
||
xce_hook_fini(&(c->hooksys));
|
||
|
||
cleanup_openssl();
|
||
|
||
job_obj_fini();
|
||
|
||
xstring_table_fini();
|
||
|
||
return;
|
||
}
|