#include "config.h" #include #include #include #include #include #include #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); } /* ha-heartbeat-committed-propagation task 2.2: after the unchanged * liveness/reply path above, a heartbeat that carries a HeartbeatMsg * body lets a standby learn the primary's committed position. The * consume helper applies the non-primary gate (Req 3.4) + term-guard * (Req 3.2) and a NULL/zero no-op (Req 4.2); a bodyless nop skips it * entirely. Purely additive — no existing OP_nop behaviour changes * (Req 4.1). */ if (request.body_which == Request_body_heartbeatmsg) ha_consume_heartbeat_committed(c, request.body.heartbeatmsg); 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_ctrlstatus: rc = handle_ctrlstatus(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; /* ha-term-safety task 1.2: candidate's Raft (lastLogTerm, lastLogIndex) * tuple drives the library logOk freshness gate (Req 2.3/2.4/2.6), * distinct from the config_tag/last_log_lsn application gate above. */ uint64_t cand_llt = vrq ? vrq->last_log_term : 0; uint64_t cand_lli = vrq ? vrq->last_log_index : 0; uint64_t local_tag = c->myconf->config_version_tag; /* Task 4.2 (ha-commit-durability): the voter floor's local-state LSN * input is the DURABLE WAL tip (persist_durable_floor), not the * volatile applied_lsn. A voter must not demand of a candidate a log * position it itself only holds volatilely, and a truncating restart * must lower this voter's floor to its real durable state. */ uint64_t local_lsn = persist_durable_floor(&c->persist); /* Election-safety fix #2 (retained): the freshness floor is the max of * our local DURABLE state and our learned-committed cluster state, so we * deny candidates that are outdated against what the cluster has already * committed — the learned_committed backstop stays on the voter axis * (bug-339 / design §1b). */ 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, cand_llt, cand_lli, &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_preVoteRequest: { /* Task 3.1 (ha-prevote): answer an incoming PreVote straw-poll. * * A PreVote asks "would you vote for me in cand_term (= sender's * current term + 1)?" WITHOUT any side effect. This case mirrors the * OPCode_requestVote case above but is strictly NON-MUTATING: * - it NEVER adopts cand_term (PreVote exists precisely to stop a * temporarily-unreachable peer from inflating the cluster term); * - it records NO vote (raft_on_pre_vote_request writes nothing); * - it never persists. * The reply therefore carries the voter's CURRENT term UNCHANGED, on * both grant and deny (Req 2.3, 6.2). * * Decision order (identical gates to the real-vote path, reused — Req * 2.1, 2.2, 3.1, 3.2, 3.3): * 1. freshness floor (ha_vote_self_freshness): deny if candidate is * below max(local, learned-committed) config_tag/LSN; * 2. raft_on_pre_vote_request: stale-term deny + the SHARED * raft_cb_vote_eligible gate (PRIMARY-deny / fresh-heartbeat * stickiness / catch-up deny). * Every denial is logged with the candidate id + reason (Req 7.2). */ 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; /* ha-term-safety task 1.2: candidate's Raft (lastLogTerm, lastLogIndex) * tuple drives the SHARED library logOk gate (identical to real-vote). */ uint64_t cand_llt = vrq ? vrq->last_log_term : 0; uint64_t cand_lli = vrq ? vrq->last_log_index : 0; uint64_t local_tag = c->myconf->config_version_tag; /* Task 4.2 (ha-commit-durability): voter-floor local-state LSN is the * DURABLE WAL tip (persist_durable_floor), not the volatile applied_lsn * — same change as the real-vote path above. */ uint64_t local_lsn = persist_durable_floor(&c->persist); /* Effective freshness floor: max of local DURABLE state and * learned-committed cluster state (election-safety fix #2 backstop * retained on the voter axis — design §1b). */ uint64_t my_tag = 0, my_lsn = 0; ha_vote_self_freshness(&c->ha, local_tag, local_lsn, &my_tag, &my_lsn); int freshness_ok = !(cand_tag < my_tag || (cand_tag == my_tag && cand_lsn < my_lsn)); /* Map candidate name → integer peer index (same loop as real-vote). */ 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; } } } /* PreVote NEVER adopts the candidate term: reply with our CURRENT term, * read once up front so grant and deny paths share it. */ uint64_t reply_term = atomic_load_explicit(&c->ha.term, memory_order_acquire); const char *deny_reason = NULL; int granted = 0; if (!freshness_ok) { deny_reason = "freshness"; } else if (from_idx < 0) { deny_reason = "unknown_candidate"; } else { /* raft_on_pre_vote_request is pure: stale-term + eligibility gate, * no term/voted_for/role mutation, no persist. */ int r = raft_on_pre_vote_request(&c->ha.raft, from_idx, cand_term, cand_llt, cand_lli); if (r == RAFT_GRANT) { granted = 1; } else { /* Distinguish the denial reason for operator diagnosis * (Req 7.2). raft_on_pre_vote_request denies on (a) stale term * — cand_term < raft.term — or (b) the shared eligibility gate * (raft_cb_vote_eligible). Mirror that gate's exact predicate * ORDER so the logged reason cannot disagree with the actual * denial: stale-term, then PRIMARY, then fresh-heartbeat on a * STANDBY, then catch-up. ha_heartbeat_is_fresh is a pure read * exposed for this purpose, so the fresh-heartbeat-vs-catch-up * overlap is reported exactly as the gate decided it. */ uint64_t my_raft_term = c->ha.raft.term; if (cand_term < my_raft_term) { deny_reason = "stale_term"; } else { ctrl_role_t role = (ctrl_role_t)atomic_load_explicit( &c->ha.role, memory_order_acquire); if (role == CTRL_PRIMARY) deny_reason = "primary"; else if (role == CTRL_STANDBY && c->ha.primary_ctrl_fd >= 0 && ha_heartbeat_is_fresh(&c->ha)) deny_reason = "fresh_heartbeat"; else if (atomic_load_explicit(&c->ha.in_catch_up, memory_order_acquire)) deny_reason = "catch_up"; else deny_reason = "eligibility"; } } } if (!granted) { INFO("HA: PreVote DENIED candidate='%s' term=%" PRIu64 " reason=%s (my_term=%" PRIu64 ", floor tag=%" PRIu64 " lsn=%" PRIu64 ")", (cand_id && cand_id[0]) ? cand_id : "", cand_term, deny_reason, reply_term, my_tag, my_lsn); } request_vote_reply_t pv_rep; memset(&pv_rep, 0, sizeof(pv_rep)); pv_rep.term = reply_term; /* UNCHANGED — no adoption */ pv_rep.vote_granted = granted; pv_rep.peer_config_tag = my_tag; pv_rep.peer_last_log_lsn = my_lsn; /* Reply over the dedicated preVoteReply opcode so the candidate's * reply router (task 5.1) feeds it to the PreVote tally, never the * real-vote tally. */ reply.opcode = OPCode_preVoteReply; reply.status = 0; reply.data_which = RequestReply_data_requestvotereplymsg; reply.data.requestvotereplymsg = &pv_rep; xce_reply(ep, c->myhost, &reply); /* Inbound PreVote connections are vote-only, like real-vote: leave the * connection open; no state push on the reply. */ 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(); /* --sync-timeout-ms override (0 = keep the persist default). */ if (c->myconf->sync_timeout_ms > 0) pcfg.sync_timeout_ms = c->myconf->sync_timeout_ms; /* --wal-sync-ms flush-cadence override (0 = keep the persist * default). Applied before persist_init so persistence starts at * the requested cadence (ha-commit-durability Req 3.1). */ persist_config_set_wal_sync_ms(&pcfg, c->myconf->wal_sync_ms); 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; } /* bug-848: a (re)started member recovered its own WAL above — make it * ADVERTISE that recovered position for catch-up (sr->applied_lsn), not the * ha_init default of 0. Otherwise the primary streams the whole log and the * receiver overwrites already-consistent records instead of fetching only * the missing suffix (Raft incremental catch-up). */ ha_sync_applied_lsn_from_recovery(&c->ha, &c->persist); INFO("HA: persist_recover done — recovered durable_floor=%" PRIu64 " current_lsn=%" PRIu64 "; applied_lsn=%" PRIu64 " advertised for catch-up", persist_durable_floor(&c->persist), (uint64_t)c->persist.current_lsn, (uint64_t)atomic_load_explicit(&c->ha.applied_lsn, memory_order_acquire)); 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) { /* c->running is the controller's liveness flag: 1 while the run loop is * active, 0 once a shutdown request breaks the loop below. check_controller() * (the streaming gate for joblist/workerlist replies in xce_reply_get) treats * !running as "shutting down" and zeroes the reply chunk. The legacy * single-node bring-up set this in cluster_start(), but cluster_start is * orphaned (not on the HA myapp_run path), so it was never set on HA nodes — * leaving joblist/workerlist permanently empty on the primary. Own the whole * flag lifecycle here, symmetric with the `c->running = 0` on shutdown. */ c->running = 1; 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; }