#define _GNU_SOURCE #include #include #include #include #include #include #include #include #include "cluster.h" #include "ha.h" #include "myconf.h" #include "netio.h" #include "persist.h" #include "delay.h" #include "request.h" #include "utils.h" #include "xce/cluster_def.h" /* forward declaration — implemented in startup_validation.c */ int validate_startup_config(const char *config_path, char *warn_buf, size_t warn_len); /* forward declaration — implemented in config_repl.c */ int load_primary_config(cluster_t *c); time_t last_print_time = 0; unsigned long npkts = 0; unsigned long npkts_old = 0; /* * myapp_run — fully non-blocking startup path (Task 5.1 / Req 1.1, 9.3, 9.4). * * Sequence: * 1. cluster_init_phase1 — hooks, managers, peer zarray * 2. SSL contexts — server ctx + client ctx for outgoing votes * 3. iomgr + vtaskmgr — init and start before election * 4. cluster_register_listener — listener in epoll before election * 5. delaymgr_init/start — required before ha_start_election_async * (election timeout watchdog uses delaymgr) * 6. persist_init/start — WAL directories + fsync thread * 7. load_primary_config — pre-load config_tag for freshness gate * 8. ha_start_election_async — non-blocking: arms watchdog, connects peers * 9. cluster_run — iomgr event loop; election resolves inside * * Removed (Req 9.3): ha_wait_for_heartbeat_or_timeout, am_i_primary_quorum, * for(;;) retry loop, cluster_init_phase2, cluster_start, * post-loop config fetching. * These are replaced by the freshness gate in voting and primary push * replication (ha_elect_evaluate_and_signal calls cluster_init_phase2 from * vtask context after the election terminal state is reached). */ int myapp_run(myconf_t *myconf) { cluster_t c; int rc; memset(&c, 0x0, sizeof(c)); c.myconf = myconf; /* ── Step 1: Phase 1 init ─────────────────────────────────────────────── */ rc = cluster_init_phase1(&c); if (rc < 0) { ERROR("cluster_init_phase1: %s", strerror(-rc)); cluster_fini(&c); return rc; } /* Wire raft_c before iomgr starts so raft_cb_persist can look up * candidate names when processing incoming vote requests during the * startup stagger nap. Without this, voted_for_name is cleared and the * already_voted check fails → unnecessary duplicate elections. */ c.ha.raft_c = &c; /* ── Step 2: SSL contexts ─────────────────────────────────────────────── */ /* Validate cert/key files early to fail fast before any network setup */ { char warn_buf[256] = {0}; validate_startup_config(myconf->conffile, warn_buf, sizeof(warn_buf)); if (warn_buf[0]) { WARN("startup: %s", warn_buf); } } initialize_openssl(); /* Server context — used for incoming connections and iomgr */ SSL_CTX *ssl_ctx = create_ssl_context(myconf->certfile, myconf->keyfile); if (!ssl_ctx) { ERROR("create_ssl_context failed"); cluster_fini(&c); return -EINVAL; } c.ha.ssl_ctx = ssl_ctx; /* Client context — used for outgoing vote connections (Req 2.2) */ SSL_CTX *client_ssl_ctx = SSL_CTX_new(TLS_client_method()); if (!client_ssl_ctx) { ERROR("SSL_CTX_new(TLS_client_method) failed"); cluster_fini(&c); return -EINVAL; } SSL_CTX_set_verify(client_ssl_ctx, SSL_VERIFY_NONE, NULL); c.ha.client_ssl_ctx = client_ssl_ctx; /* ── Step 3: iomgr + vtaskmgr ────────────────────────────────────────── */ /* Req 1.2: open the listener socket before any election probe so that * concurrent-startup peers can reach this node's vote handler. */ rc = cluster_open_listener(&c); if (rc < 0) { ERROR("cluster_open_listener: %s", strerror(-rc)); cluster_fini(&c); return rc; } /* Req 1.2: init and start iomgr/vtaskmgr before election so that incoming * OP_requestVote RPCs are dispatched during the election phase. */ iomgr_init(&(c.iomgr)); iomgr_set_ssl_ctx(&(c.iomgr), ssl_ctx); { int nvcpus = (myconf->vcpus > 0) ? myconf->vcpus : 8; vtaskmgr_init(&(c.vtaskmgr), nvcpus); } iomgr_start(&(c.iomgr)); vtaskmgr_start(&(c.vtaskmgr)); /* ── Step 4: register listener ───────────────────────────────────────── */ /* Register listener with epoll before election so iomgr can accept and * dispatch incoming OP_requestVote connections. */ { int monitor_rc = cluster_register_listener(&c); if (monitor_rc < 0) { ERROR("cluster_register_listener: %s", strerror(-monitor_rc)); cluster_fini(&c); return monitor_rc; } } /* ── Step 5: delaymgr ────────────────────────────────────────────────── */ /* Init and start delaymgr before ha_start_election_async. * The election timeout watchdog uses delay_task_ms() which requires * delaymgr to be running (Req 1.3 non-blocking election watchdog). * Pass NULL runner_fn → delaymgr uses its built-in default runner. */ delaymgr_init(&(c.delaymgr), 200 /* precision_ms */, NULL); delaymgr_start(&(c.delaymgr)); /* ── Step 6: persist ─────────────────────────────────────────────────── */ /* Init persist before election so WAL writes are possible as soon as * cluster_init_phase2 runs from vtask context after election result. */ { const char *workdir = myconf->workdir ? myconf->workdir : DEFAULT_WORKDIR; const char *name = myconf->name ? myconf->name : "controller"; persist_config_t pcfg = persist_config_defaults(); persist_set_workdir(&c.persist, workdir, name); rc = persist_init(&c.persist, &pcfg, &c); if (rc != 0) { ERROR("persist_init: %s", strerror(-rc)); cluster_fini(&c); return rc; } } rc = persist_start(&c.persist); if (rc != 0) { ERROR("persist_start: %s", strerror(-rc)); cluster_fini(&c); return rc; } /* ── Step 7: pre-load config_tag ─────────────────────────────────────── */ /* Pre-load config from disk if a local config file exists, so that this * node's config_tag is known before election. The freshness gate in * handle_request_vote (Req 4.2) requires config_tag to be set here so that * a node with a valid local config wins over nodes with no config (tag=0). * If no config file exists yet, config_version_tag stays 0 and the node * will receive the config via the primary's replication push post-election * (Req 7.3). */ { int pre_rc = load_primary_config(&c); if (pre_rc == 0) { INFO("Pre-election config loaded: tag=%016llx", (unsigned long long)c.myconf->config_version_tag); } /* Ignore errors — node may have no config yet (first boot or standby) */ } /* ── Step 8: ha_start_election_async ─────────────────────────────────── */ /* Build peer pointer array from peers_cfg. * n_peers = total cluster size (including self). * peers[] = array of OTHER candidates (excluding self, identified by * ha->my_priority). ha_start_election_async handles single-node * shortcircuit when n_peers == 1. */ const xce_cluster_config_t *pcfg = &myconf->peers_cfg; int n_peers = (pcfg->candidates_count > 0) ? pcfg->candidates_count : 1; xce_candidate_t *peer_ptrs[MAX_HA_CANDIDATES]; int n_other = 0; for (int i = 0; i < pcfg->candidates_count; i++) { if (i != c.ha.my_priority) { peer_ptrs[n_other++] = (xce_candidate_t *)&pcfg->candidates[i]; } } uint64_t my_tag = myconf->config_version_tag; uint64_t my_lsn = (uint64_t)atomic_load_explicit(&c.ha.applied_lsn, memory_order_acquire); /* Initialize raft state machine once with cluster topology. * Must happen before any election or watchdog path that calls * raft_on_leader_lost() / raft_start_election(). */ ha_init_raft(&c.ha, &c, n_peers); /* Startup stagger: delay election start by priority * election_priority_step_ms. * Node 0 starts first; lower-priority nodes wait, see the winner's vote request * arrive, and grant the vote instead of starting their own election. * Uses the same step as the existing blocking-election probe path (Req 1.2). */ if (c.ha.my_priority > 0) { uint32_t step_ms = c.ha.config.election_priority_step_ms ? c.ha.config.election_priority_step_ms : HA_DEFAULT_ELECTION_PRIORITY_STEP_MS; long delay_ns = (long)c.ha.my_priority * (long)step_ms * 1000000L; struct timespec jitter = { .tv_sec = delay_ns / 1000000000L, .tv_nsec = delay_ns % 1000000000L, }; nanosleep(&jitter, NULL); } /* If a leader is already known after the startup stagger (either because * iomgr processed an incoming vote request and updated the raft term, or * because ctrlhello arrived on a promoted vote ep), skip starting our own * election. The voted_for_name check catches the synchronous vote-grant * path (iomgr was already running); the raft term + primary_ctrl_fd checks * catch the case where iomgr was NOT running during the nap (connections * queued in the kernel backlog) but the winner's ctrlhello has already * established the control channel. * * voted_for_name is combined with term: a self-vote means this node was * a candidate, not a follower — no external leader is known. In * production raft_cb_persist records the self-vote under this node's own * name (myconf->name); the "self" placeholder only appears in test mode, * so both spellings must be checked. A stale vote (voted_for_term < * ha->term) is also discarded. All these cases must fall through to * ha_start_election_async. */ { int leader_known = 0; pthread_mutex_lock(&c.ha.lock); uint64_t vft = c.ha.voted_for_term; uint64_t cur = atomic_load_explicit(&c.ha.term, memory_order_relaxed); if (c.ha.voted_for_name[0] != '\0' && strcmp(c.ha.voted_for_name, "self") != 0 && strcmp(c.ha.voted_for_name, myconf->name) != 0 && vft == cur) leader_known = 1; pthread_mutex_unlock(&c.ha.lock); if (!leader_known && c.ha.raft.term > 0) leader_known = 1; if (!leader_known && c.ha.primary_ctrl_fd >= 0) leader_known = 1; if (leader_known) { INFO("myapp_run: leader already known after startup delay " "(voted_for=%s, voted_term=%" PRIu64 ", raft_term=%" PRIu64 ", primary_fd=%d) — skipping own election", c.ha.voted_for_name[0] ? c.ha.voted_for_name : "(none)", vft, c.ha.raft.term, c.ha.primary_ctrl_fd); goto skip_election; } } /* Non-blocking: arms election timeout watchdog, connects outgoing vote * endpoints, returns immediately. Election resolves via iomgr callbacks; * ha_elect_evaluate_and_signal calls cluster_init_phase2 from vtask context * and sets ha->role atomically. ha_dispatch_gate opens when role is set. */ ha_start_election_async(&c.ha, &c, n_peers, (n_other > 0) ? peer_ptrs : NULL, my_tag, my_lsn, myconf->name); skip_election:; /* All standbys wait passively for the primary to connect outbound and push * config via ctrlhello. fetch_config_from_primary has been removed: * standbys never initiate outbound connects after an election result. */ { ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&c.ha.role, memory_order_acquire); if (role == CTRL_STANDBY || role == CTRL_INIT) { atomic_store_explicit(&c.ha.role, (int32_t)CTRL_STANDBY, memory_order_release); INFO("myapp_run: CTRL_STANDBY — waiting for primary to connect and push config"); /* Arm heartbeat watchdog even when we skipped the election. * If the known primary never connects within heartbeat_timeout_ms, * the watchdog fires ha_on_primary_ctrl_fd_rdhup which increments * the term (clearing voted_for) and starts a fresh election. */ ha_arm_heartbeat_timeout_vtask(&c.ha, &c); } } /* ── Step 9: cluster_run ─────────────────────────────────────────────── */ /* Req 9.4: myapp_run exits into cluster_run within 10ms of startup. * cluster_run is the iomgr event loop — just waits for stop signal. * All election I/O and role transitions happen inside the loop. */ rc = cluster_run(&c); cluster_fini(&c); myconf_fini(myconf); return rc; }