#include "config.h" #include #include #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 finalizes ep via endpoint_killer (now endpoint_unref). */ zarray_del(&(c->endpoints), ep->idx); return; } /* bug-1298: not in c->endpoints — drop the owner ref (deferred fini+free if a * pool vtask still references it) instead of freeing inline on the io_worker. */ endpoint_unref(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. Skipped when the entry's * own term stamp proves it is already stale (bug-1418 family): a * stale-term AppendEntries is about to be rejected by the raft library * below anyway, and letting it ALSO reset liveness would mean a standby * that legitimately outraced its own peer to a higher term (a rare but * valid election-timing outcome) never times out to start a fresh, * higher-term election — permanently deadlocking the cluster on a * primary whose term the standby correctly knows is stale. */ if (!ha_msg_term_is_stale(&c->ha, entry->term)) { 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; } /* ---- OPCode_schedAdmin: forward-and-block-for-reply support (task 10.1) -- * * The controller is a stateless pass-through for OPCode_schedAdmin: it * forwards a SchedAdminRequest to the scheduler over c->sched.ep and must * relay the scheduler's SchedAdminReply back to the original caller, which * means the request-handling vtask thread has to block until that reply * arrives (or 5s elapses) -- unlike every other controller->scheduler call * in this file (sched.c's xce_push_*_req / xce_pushconfig_req), which are * fire-and-forget. * * This mirrors on_fetchfile's proxy pattern (job.c): register a waiter * BEFORE sending (avoids a race with a reply that arrives before the wait * starts), send async via the normal xce_request() path, then block on a * condvar with a deadline. The one difference from fetchfile's keyed * (jobid) multi-waiter list: this needs only a SINGLE slot, because the * schedAdmin case below holds c->sched.lock for the entire forward+wait * duration, which already serializes at most one in-flight schedAdmin * round-trip per controller at a time (an accepted trade-off -- xceadmin * admin commands are low-frequency, human-paced traffic, not a hot path; * see design research for task 10.1). */ typedef struct { pthread_mutex_t mu; pthread_cond_t cv; int done; int status; sched_admin_reply_t* reply; } schedadmin_pending_t; static pthread_mutex_t g_schedadmin_mu = PTHREAD_MUTEX_INITIALIZER; static schedadmin_pending_t* g_schedadmin_pending = NULL; static void schedadmin_register(schedadmin_pending_t* p) { pthread_mutex_lock(&g_schedadmin_mu); g_schedadmin_pending = p; pthread_mutex_unlock(&g_schedadmin_mu); } static void schedadmin_unregister(schedadmin_pending_t* p) { pthread_mutex_lock(&g_schedadmin_mu); if (g_schedadmin_pending == p) g_schedadmin_pending = NULL; pthread_mutex_unlock(&g_schedadmin_mu); } /* Signal the pending schedAdmin waiter (if any) with the scheduler's reply. * Called from the OPCode_reply case below when a SchedAdminReply arrives on * c->sched.ep. Returns 1 if a waiter was found and signaled (ownership of * `sar` transfers to it -- the caller must NULL its own copy so the normal * xce_free_request path does not also free it), 0 if no waiter was * registered (e.g. a late reply arriving after this controller already gave * up and replied "request timeout" to xceadmin -- `sar` stays owned by the * caller in that case, freed normally). */ static int schedadmin_signal(int status, sched_admin_reply_t* sar) { pthread_mutex_lock(&g_schedadmin_mu); schedadmin_pending_t* p = g_schedadmin_pending; if (p == NULL) { pthread_mutex_unlock(&g_schedadmin_mu); return 0; } pthread_mutex_lock(&p->mu); p->status = status; p->reply = sar; p->done = 1; pthread_cond_signal(&p->cv); pthread_mutex_unlock(&p->mu); pthread_mutex_unlock(&g_schedadmin_mu); return 1; } #ifdef XCE_TESTING /* * schedadmin_set_timeout_ms_for_test / SCHEDADMIN_TIMEOUT_MS (task 10.2, Req * 5.10 edge case "Scheduler disconnects mid-request"): the OPCode_schedAdmin * case's forward-and-block-for-reply deadline is a fixed 5s in production * (requirements.md AC 10: "await scheduler reply (timeout: 5s)"). Proving * that deadline genuinely expires needs a REAL pthread_cond_timedwait * timeout to fire -- not a shortcut that bypasses the wait -- but a real 5s * wait per test run has no precedent in this suite (the longest real waits * elsewhere are sub-second, e.g. DEFAULT_SYNC_TIMEOUT_MS=500ms in * test_commit_resolve.c). This test-only override shrinks the deadline so * the SAME wait-then-give-up code path below executes for real, just * faster. Guarded by XCE_TESTING like myreq_handler_for_test below -- never * compiled into production builds, where SCHEDADMIN_TIMEOUT_MS is a bare * 5000 with zero indirection and the deadline math below folds to exactly * the prior `deadline.tv_sec += 5` (5000 / 1000 == 5, 5000 % 1000 == 0). */ static long g_schedadmin_timeout_ms = 5000; void schedadmin_set_timeout_ms_for_test(long ms) { g_schedadmin_timeout_ms = ms; } #define SCHEDADMIN_TIMEOUT_MS g_schedadmin_timeout_ms #else #define SCHEDADMIN_TIMEOUT_MS 5000L #endif 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); /* bug-1286: a chaotic-restart race can run this handler on an inbound endpoint * whose owner (cluster_t) is NULL — not yet wired at accept time, or cleared * during teardown. With c == NULL, `&c->ha` is a NULL+offset garbage pointer * (0x3ff0) that SIGSEGV'd ha_dispatch_gate (ha.c) when it loaded ha->role. * Drop the request LOUDLY instead of dereferencing it: we have no cluster * context to dispatch through and cannot even address a reply (it needs * c->myhost). The controller stays alive (never self-terminate on an * operational condition); the peer retries once the endpoint is owned. The * request is freed here so the decoded packet does not leak. */ if (c == NULL) { WARN("myreq_handler: endpoint %p has no owner (cluster) — dropping %s " "(restart race, bug-1286)", (void*)ep, xce_request_op_name(&request)); xce_free_request(&request); return -1; } /* 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. * Skipped when the heartbeat's own term stamp proves it is already * stale (bug-1418 family) -- see the identical guard's doc comment * in handle_wal_catchup_entry above. A bodyless nop (no HeartbeatMsg) * carries no term, so ha_msg_term_is_stale's msg_term==0 case * preserves the existing unconditional behavior for it. */ { uint64_t hb_term = (request.body_which == Request_body_heartbeatmsg && request.body.heartbeatmsg) ? request.body.heartbeatmsg->term : 0; if (!ha_msg_term_is_stale(&c->ha, hb_term)) { 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_fetchfile: rc = on_fetchfile(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) { /* bug-1435: serialize against concurrent raft_state_t access — * this node's own vote-reply/election processing (a different * peer connection, dispatched on a different pool worker) can * race this vote REQUEST on the same ha->raft (see ha.h * raft_lock doc). */ pthread_mutex_lock(&c->ha.raft_lock); int r = raft_on_vote_request(&c->ha.raft, from_idx, cand_term, cand_llt, cand_lli, &reply_term); pthread_mutex_unlock(&c->ha.raft_lock); 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. Still reads * raft_state_t's term/role — bug-1435: take raft_lock so this * read cannot observe a torn (mid-transition) snapshot while a * concurrent vote-reply/election-start call on another pool * worker is mutating the same fields (see ha.h raft_lock doc). */ pthread_mutex_lock(&c->ha.raft_lock); int r = raft_on_pre_vote_request(&c->ha.raft, from_idx, cand_term, cand_llt, cand_lli); pthread_mutex_unlock(&c->ha.raft_lock); 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 an admin client (or xceadmin's * config-push tool) TO the primary controller, which decodes the * blob, verifies it, and drives it through config_commit_change() * (the (term,lsn) marker consensus path) — see * do_upgrade_config_with_bytes's doc comment, config_repl.c. Only the * primary can commit a WAL entry, so a standby cannot service this * request; ha_dispatch_gate already denies OPCode_upgradeConfig for * every non-primary role (it is absent from the CTRL_STANDBY / * CTRL_INIT / CTRL_ELECTING allow-lists, ha.c), so this check is a * defensive belt-and-suspenders guard, not the primary gate. * * bug-fix (2026-07-25): this check was inverted from * commit fd93997f ("election-iomgr-refactor", 2026-05-31) onward — * it rejected the request WHEN primary, unconditionally reachable * only on primary (per the dispatch-gate note above), making * handle_upgrade_config() permanently unreachable from any network * caller (only unit tests calling it directly ever exercised it). * Independently diagnosed and documented as a blocking risk by the * admin-config-tool spec (design.md "Boundary Commitments" / * "Risks") before config-upgrade-snapshot task 8.6 needed a live * OP_upgradeConfig sender and hit the same wall. Restored to the * pre-regression (correct) polarity. */ if (!ha_is_primary(&c->ha)) { /* Standby cannot commit a config change — reject loud. */ 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_schedAdmin: { /* Stateless pass-through forward to the scheduler (task 10.1, Req * 5.10). The controller commits no WAL entry and interprets no * field beyond the opcode -- every field of the request/reply * passes through untouched. Structurally mirrors OPCode_upgradeConfig * above: primary-only gate, explicit reply on every path, never a * dropped request. See the schedadmin_* helpers above this function * for the forward-and-block-for-reply mechanism (mirrors * on_fetchfile's proxy pattern in job.c). */ sched_admin_req_t* areq = request.body.schedadminmsg; /* Defensive belt-and-suspenders guard -- mirrors OPCode_upgradeConfig's * identical inner check above. The real gate is ha_dispatch_gate * (called before this switch, ~line 276): OPCode_schedAdmin is * absent from the CTRL_STANDBY / CTRL_INIT / CTRL_ELECTING * allow-lists in ha.c -- exactly like OPCode_upgradeConfig -- so a * non-primary controller is rejected with no forwarding attempt * before this switch is ever reached; this check only guards a * role change racing between that gate and here. */ if (!ha_is_primary(&c->ha)) { reply.status = -EAGAIN; reply.data_which = RequestReply_data_nodata; xce_reply(ep, c->myhost, &reply); rc = -EAGAIN; break; } if (areq == NULL) { reply.status = -EINVAL; reply.data_which = RequestReply_data_nodata; xce_reply(ep, c->myhost, &reply); rc = -EINVAL; break; } /* c->sched.lock is held for the FULL forward+wait duration below -- * an accepted trade-off (task 10.1 design notes): xceadmin admin * commands are low-frequency, human-paced traffic, not a hot path. * This also naturally serializes at most one in-flight schedAdmin * round-trip at a time, which the single-slot pending waiter above * relies on. */ mylock_lock(&(c->sched.lock)); if (c->sched.ep == NULL) { sched_admin_reply_t nc_reply; memset(&nc_reply, 0, sizeof(nc_reply)); nc_reply.error = "scheduler not connected"; mylock_unlock(&(c->sched.lock)); reply.status = 0; reply.data_which = RequestReply_data_schedadminreplymsg; reply.data.schedadminreplymsg = &nc_reply; xce_reply(ep, c->myhost, &reply); rc = 0; break; } schedadmin_pending_t pending = { .mu = PTHREAD_MUTEX_INITIALIZER, .cv = PTHREAD_COND_INITIALIZER, .done = 0, .status = -ETIMEDOUT, .reply = NULL }; /* Register before sending -- avoid a race with a reply that arrives * before pthread_cond_timedwait below starts waiting (same * discipline as fetchfile_register, job.c). */ schedadmin_register(&pending); rc = xce_schedadmin_req(c->sched.ep, areq); if (rc < 0) { schedadmin_unregister(&pending); mylock_unlock(&(c->sched.lock)); reply.status = rc; reply.data_which = RequestReply_data_nodata; xce_reply(ep, c->myhost, &reply); break; } struct timespec deadline; long schedadmin_timeout_ms = SCHEDADMIN_TIMEOUT_MS; clock_gettime(CLOCK_REALTIME, &deadline); deadline.tv_sec += schedadmin_timeout_ms / 1000; deadline.tv_nsec += (schedadmin_timeout_ms % 1000) * 1000000L; if (deadline.tv_nsec >= 1000000000L) { deadline.tv_sec += 1; deadline.tv_nsec -= 1000000000L; } pthread_mutex_lock(&pending.mu); while (!pending.done) { int wrc = pthread_cond_timedwait(&pending.cv, &pending.mu, &deadline); if (wrc == ETIMEDOUT) break; } pthread_mutex_unlock(&pending.mu); /* Unregister immediately after releasing pending.mu, BEFORE reading * the outcome below -- closes the window where a late-arriving * reply could still find this slot registered and hand off `sar` * ownership into it after we have already stopped looking (the * abandoned-slot leak this fix addresses). * * This must NOT be done while still holding pending.mu: * schedadmin_unregister() only takes g_schedadmin_mu (confirmed by * its body above -- it never touches p->mu), but schedadmin_signal() * takes g_schedadmin_mu THEN p->mu. Calling unregister while holding * pending.mu would acquire the two locks in the reverse order * (pending.mu then g_schedadmin_mu) from a concurrent * schedadmin_signal() (g_schedadmin_mu then pending.mu) -- a classic * AB-BA lock-ordering inversion that can deadlock. Releasing * pending.mu first keeps both threads' lock order consistent * (g_schedadmin_mu is always acquired without holding pending.mu). * * Once schedadmin_unregister() returns, no NEW signal can find this * slot. Re-acquiring pending.mu below picks up the result of any * signal that was already in flight (had read g_schedadmin_pending * before we cleared it) and finished its write before we got here -- * mutual exclusion on g_schedadmin_mu guarantees that race resolves * one way or the other before schedadmin_unregister() returns. */ schedadmin_unregister(&pending); pthread_mutex_lock(&pending.mu); int got_done = pending.done; int got_status = pending.status; sched_admin_reply_t* got_reply = pending.reply; pthread_mutex_unlock(&pending.mu); if (got_done && got_reply != NULL) { /* Relay the scheduler's SchedAdminReply back to xceadmin * verbatim -- every field passes through untouched. */ reply.status = 0; reply.data_which = RequestReply_data_schedadminreplymsg; reply.data.schedadminreplymsg = got_reply; xce_reply(ep, c->myhost, &reply); free_SchedAdminReply_ptr(&got_reply); } else if (got_done) { /* Scheduler replied at the RPC level with no SchedAdminReply * payload (unexpected, but handled loudly, never dropped) -- * relay its RPC status. */ reply.status = got_status; reply.data_which = RequestReply_data_nodata; xce_reply(ep, c->myhost, &reply); } else { sched_admin_reply_t to_reply; memset(&to_reply, 0, sizeof(to_reply)); to_reply.error = "request timeout"; reply.status = 0; reply.data_which = RequestReply_data_schedadminreplymsg; reply.data.schedadminreplymsg = &to_reply; xce_reply(ep, c->myhost, &reply); } mylock_unlock(&(c->sched.lock)); rc = 0; break; } case OPCode_reply: { /* Reply arriving on an inbound endpoint. Three independent producers * feed this case: * 1. fetchfile proxy replies from a worker (fetchfile feature) — * routed to the pending waiter via fetchfile_signal. * 2. Config-push ACKs 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). Fed 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). * 3. SchedAdminReply arriving on c->sched.ep in response to the * OPCode_schedAdmin case's forwarded request — routed to the * pending waiter via schedadmin_signal (task 10.1). */ xce_reply_t* workreply = request.body.replymsg; if (workreply != NULL && workreply->opcode == OPCode_fetchfile) { xce_fetchfile_reply_t* fr = (workreply->data_which == RequestReply_data_fetchfilereply) ? workreply->data.fetchfilereply : NULL; uint64_t jid = fr ? fr->jobid : 0; fetchfile_signal(jid, workreply->status, fr); /* Ownership of fr transfers to the pending waiter. */ workreply->data.fetchfilereply = NULL; rc = 0; break; } if (workreply != NULL && workreply->opcode == OPCode_schedAdmin) { sched_admin_reply_t* sar = (workreply->data_which == RequestReply_data_schedadminreplymsg) ? workreply->data.schedadminreplymsg : NULL; if (schedadmin_signal(workreply->status, sar)) { /* Ownership of sar (if any) transferred to the pending * waiter — NULL it so xce_free_request below does not also * free it. If no waiter was found (e.g. this reply arrived * after the forwarder already gave up at the 5s timeout), * leave it owned by workreply for the normal free path. */ workreply->data.schedadminreplymsg = NULL; } rc = 0; break; } if (request.body_which == Request_body_replymsg && workreply != NULL && workreply->data_which == RequestReply_data_replconfigpushreplymsg && workreply->data.replconfigpushreplymsg) { repl_config_push_reply_t *cack = workreply->data.replconfigpushreplymsg; /* raft-adapter-conformance Req 7.4/7.5: a config-push reply now * carries the replying member's current term (task 1 schema * addition + task 2.1 send side) -- route it through the same * term-discovery entry point the heartbeat-reply and WAL-NACK * vectors already use (bug-1421; ha.c raft_lock-guarded * raft_on_higher_term_observed calls) so a stale-term primary * adopts the higher term and steps down instead of retrying * forever against a standby that has already moved on. Applies * regardless of accept/reject below -- a reject can still carry * a useful higher term. raft_on_higher_term_observed is itself * strictly-greater-than-only (ha-async-io.md Heartbeat Model), * so no separate `>` comparison is needed here; raft_lock * discipline mirrors the vote-request/pre-vote-request branches * above (bug-1435 -- serialize against concurrent raft_state_t * access). */ pthread_mutex_lock(&c->ha.raft_lock); (void)raft_on_higher_term_observed(&c->ha.raft, cack->term); pthread_mutex_unlock(&c->ha.raft_lock); 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; } #ifdef XCE_TESTING /* * myreq_handler_for_test: expose myreq_handler for tests (peer_ctrl_msg_ * handler_for_test precedent, ha.c). Builds the vtask_t/vtask_data_t wrapper * vtask_entry constructs internally in production so a test can decode+ * dispatch a real packet through the actual switch -- including the * ReplConfigPushReply branch's term-adoption check (raft-adapter- * conformance Req 7.4/7.5). Packet ownership follows myreq_handler's own * contract: myreq_handler unconditionally calls xce_decode(pkt, ...) as its * first step, and xce_decode() itself consumes pkt -- it calls * free_packet(pkt) (frees pkt->data) and FREEUP(pkt) (frees the io_packet_t * struct) before returning (request.c). Ownership of `pkt` is therefore * TRANSFERRED into this call: the caller must NOT touch or free `pkt` after * calling myreq_handler_for_test -- doing so is a use-after-free/double-free. * (xce_free_request separately releases the decoded xce_request_t, an * unrelated stack-local object in myreq_handler.) */ int myreq_handler_for_test(endpoint_t* ep, io_packet_t* pkt) { vtask_data_t data; memset(&data, 0, sizeof(data)); data.vtype = VTASK_IO; data.u.iodata.ep = ep; data.u.iodata.pkt = pkt; vtask_t task; memset(&task, 0, sizeof(task)); task.data = &data; return myreq_handler(&task); } #endif /* XCE_TESTING */ 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_unref(ep2); /* bug-1298: refcnt 1->0, fini+free (no queued vtasks yet) */ 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_unref(ep2); /* bug-1298: refcnt 1->0, fini+free (not yet queued) */ 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; } /* bug-1298: drop the creator/owner reference instead of freeing inline. If * a pool vtask still references this ep (the io_worker error path runs while * the dispatcher may be reading ep->busy), the fini+free is deferred to the * last endpoint_unref — never racing the dispatcher. */ endpoint_unref(ep); return; } int cluster_build_config(cluster_t* c) { char valbuf[128]; 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); 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). * * Req 6.2: each tunable is sourced from its CLI-set myconf_t field * (--jobmgr-nhash etc., populated in main.c) when nonzero; a field left at * 0 (unset) falls back to the hardcoded default below, so omitting all * tunable flags reproduces today's exact values. The cluster_config * @13–@17 fields this TODO used to reference were removed entirely by * clusterconfig-schema-split Requirement 2 (they were already dead) and * will never be wired — CLI flags are the sole configuration path now. */ /* * 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. * Req 6.2: same 0-sentinel idiom as myconf->vcpus (myapp.c) — a nonzero * CLI-set myconf_t value overrides the hardcoded default; 0 (unset) * reproduces today's exact value. */ myconf_t *myconf = c->myconf; c->tunables.jobmgr_nhash = (myconf->jobmgr_nhash > 0) ? myconf->jobmgr_nhash : 512; c->tunables.jobmgr_delta = (myconf->jobmgr_delta > 0) ? myconf->jobmgr_delta : 128; c->tunables.workermgr_nhash = (myconf->workermgr_nhash > 0) ? myconf->workermgr_nhash : 128; c->tunables.workermgr_delta = (myconf->workermgr_delta > 0) ? myconf->workermgr_delta : 32; c->tunables.qmgr_nhash = (myconf->qmgr_nhash > 0) ? myconf->qmgr_nhash : 128; c->tunables.qmgr_delta = (myconf->qmgr_delta > 0) ? myconf->qmgr_delta : 32; c->tunables.resmgr_nhash = (myconf->resmgr_nhash > 0) ? myconf->resmgr_nhash : 128; c->tunables.resmgr_delta = (myconf->resmgr_delta > 0) ? myconf->resmgr_delta : 32; /* resmgr_inst_delta (myconf/CLI name) maps by MEANING, not by name, to * tunables.resmgr_inst (runtime field name) — both are the same * zarray_init() growth-delta for c->sharedinst (confirmed: zarray_init's * 2nd parameter is literally named `delta` in zarray.c). Do not rename * tunables.resmgr_inst itself — out of scope, other readers unverified. */ c->tunables.resmgr_inst = (myconf->resmgr_inst_delta > 0) ? myconf->resmgr_inst_delta : 64; c->tunables.xstrings_nhash = (myconf->xstrings_nhash > 0) ? myconf->xstrings_nhash : 512; /* myconf->xstrings_delta (CLI: --xstrings-delta, added in task 6.1) is * intentionally NOT wired to any c->tunables field: xstring_table_init() * (src/lib/xstring.c) takes only an nslots parameter, and its backing * store (kvtab_type, src/lib/kv.c) has no delta/growth-step concept at * all -- each bucket entry is calloc'd individually, unlike the * xkvmgr_type-backed jobmgr/workermgr/qmgr/resmgr tables which do grow * by a configurable delta. Adding a tunables.xstrings_delta field would * be dead: nothing could ever read it. WARN so the flag isn't a silent * no-op for an operator who sets it. */ if (myconf->xstrings_delta > 0) { WARN("--xstrings-delta=%u has no effect: the xstrings table has no " "growth-delta dimension (xstring_table_init() takes only " "nslots); use --xstrings-nhash instead", myconf->xstrings_delta); } } 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; } /* Hardcoded fallback listen port, used only when --peers was never passed * at all (true single-node mode, peers_cfg.candidates_count == 0). An * operator who wants a non-default single-node port passes a single * self-referencing --peers =: entry instead, which * resolves via the candidate-match branch below, not this constant. * Matches the pre-schema-split ClusterConfig.port default value. */ #define XCE_CONTROLLER_DEFAULT_PORT 9990 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: * 1. Search --peers candidates for the local name. * - Matched with a nonzero port: bind to it (unchanged happy * path; also how a single-node operator gets a custom port, * via a single self-referencing --peers entry). * - Matched with port == 0 (malformed --peers entry): falls * through to the hardcoded default below, same as today's * tolerated case. * - Not matched at all, but --peers WAS passed (candidates_count * > 0): likely operator misconfiguration (own name typo'd or * omitted) -- refuse to start, unchanged from today. * 2. --peers never passed at all (candidates_count == 0): true * single-node mode, use the hardcoded default. Always succeeds. */ uint32_t my_port = 0; bool matched = false; { 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; matched = true; break; } } if (!matched && pcfg->candidates_count > 0) { ERROR("cluster_open_listener: local name '%s' not found in " "--peers list; refusing to start", c->myconf->name); return -EINVAL; } } if (my_port == 0) { my_port = XCE_CONTROLLER_DEFAULT_PORT; } 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) { /* Normal during cold boot: a standby reaches phase2 (post-election) * before the primary's config-push has delivered bytes yet. Not a * divergence or a hard failure — matches load_primary_config's own * "will fetch from primary after election" INFO-level framing for * the same underlying condition. The caller retries. */ INFO("cluster_init_phase2: cluster_config not loaded yet — normal " "during cold boot while awaiting the primary's config-push; " "will retry once it arrives"); 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); /* --max-wal-records checkpoint-trigger override (0 = keep the * persist default). raft-ha-core-safety task 5.10a / ha-entity- * snapshots task 7.1: lets a soak force real checkpoints on a * deterministic record count instead of the 10000-record default. */ persist_config_set_max_wal_records(&pcfg, c->myconf->max_wal_records); /* --wal-segment-bytes physical-segment-size override (0 = keep the * persist default). raft-ha-core-safety task 5.10a / ha-entity- * snapshots task 7.1: a segment is only trimmable once it rotates * out, so forcing a real snapshot-triggered catch-up in the soak * needs a small deterministic segment size instead of the 64 MB * production default. */ persist_config_set_wal_segment_bytes(&pcfg, c->myconf->wal_segment_bytes); 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); /* raft-drive-async task 1.1: resolve and log the peer-scaled liveness * reservation at init — both inputs (n_peers, n_vcpus) are final here. * Single source of truth: task 1.6 (LivenessAdmission) consumes this * value, it does NOT recompute the clamp/WARN. */ int liveness_reserved = drive_async_liveness_reserved(c->ha.n_peers, c->vtaskmgr.n_vcpus); INFO("drive-async: VTASK_LIVENESS_RESERVED=%d (n_peers=%d, n_vcpus=%d; " "commit/REQ_BLOCK cap=%d)", liveness_reserved, c->ha.n_peers, c->vtaskmgr.n_vcpus, c->vtaskmgr.n_vcpus - liveness_reserved); /* raft-drive-async task 1.6: install the pool-level LivenessAdmission * reservation and fold commit-wait admission onto it. The pool now owns * the single reservation; commit_submit(WAIT) acquires a pool blocking * slot (cap = n_vcpus - liveness_reserved) instead of the local * COMMIT_WAIT_RESERVED cap. Dormant beyond the commit cap-value change * until peer_ctrl dispatches classified tasks (task 3.1). */ vtaskmgr_set_liveness_reserved(&c->vtaskmgr, liveness_reserved); commit_set_pool(&c->commit, &c->vtaskmgr); /* raft-drive-async task 3.1: give ha the pool handle so * peer_ctrl_msg_handler can dispatch peer-control messages onto it when * the dispatch flag is ON. Dormant until the flag flips (default OFF); * stored as void* (ha.h does not pull vtaskmgr.h). */ c->ha.pool = &c->vtaskmgr; if (c->vtaskmgr.n_vcpus < c->ha.n_peers) { /* Under-provisioned: the n_vcpus ceiling clamped the reservation * below the ideal n_peers-1. Still correct (deadlock-free via fast * worker cycling), but operator-visible per CLAUDE.md fail-loud. * A controller NEVER self-terminates on this condition. */ WARN("drive-async: under-provisioned controller — n_vcpus=%d < " "n_peers=%d; liveness reservation clamped to %d (ideal %d). " "Per-peer liveness latency may rise under load; consider " "raising --vcpus. Cluster remains correct and stays up.", c->vtaskmgr.n_vcpus, c->ha.n_peers, liveness_reserved, c->ha.n_peers - 1); } } /* raft-drive-async task 1.2: bring up the SlowLane executor — the dedicated * bounded-thread offload lane for long blocking I/O (snapshot transfer, * config-push, peer reconnect) that must never run on the iomgr/delaymgr * drive threads. Dormant infra here: no production caller until the * reconnect (2.x) / snapshot/config-push (3.x) offloads land. Sized from * the DEFAULT_SLOWLANE_* constants (task 1.1, config.h). Guard c->slowlane * so the legacy two-phase path (phase2 can run twice) inits exactly once. */ if (c->slowlane.workers == NULL) { rc = slowlane_init(&c->slowlane, DEFAULT_SLOWLANE_WORKERS, DEFAULT_SLOWLANE_MAX_INFLIGHT); if (rc != 0) { ERROR("slowlane_init failed: %d (workers=%d, max_inflight=%d)", rc, DEFAULT_SLOWLANE_WORKERS, DEFAULT_SLOWLANE_MAX_INFLIGHT); return rc; } INFO("drive-async: slowlane up (workers=%d, max_inflight=%d)", DEFAULT_SLOWLANE_WORKERS, DEFAULT_SLOWLANE_MAX_INFLIGHT); } /* 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); /* bug-1042: also raise the raft LIBRARY's log_len/commit_index to the * recovered position. ha_init_raft ran BEFORE this recovery on the standby- * rejoin boot, so without this the library stays at log_len=0 and rejects * every AppendEntries (logOk fail) — wedging the rejoined member forever. */ ha_sync_raft_log_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); /* raft-drive-async task 1.2: drain + join the SlowLane after the drive * threads have stopped submitting (ha_fini above) and before persist/iomgr * teardown, since lane tasks (snapshot/reconnect, once wired) touch those. * No-op if phase2 never ran (workers==NULL guard inside slowlane_fini). */ slowlane_fini(&c->slowlane); /* 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; }