diff --git a/.kiro/specs/election-iomgr-refactor/tasks.md b/.kiro/specs/election-iomgr-refactor/tasks.md index 9f3f645f..454cbc57 100644 --- a/.kiro/specs/election-iomgr-refactor/tasks.md +++ b/.kiro/specs/election-iomgr-refactor/tasks.md @@ -82,7 +82,7 @@ - _Boundary: ha.c (election evaluation)_ - _Depends: 2.4, 3.1_ -- [ ] 4.2 Implement config and WAL replication push to standbys +- [x] 4.2 Implement config and WAL replication push to standbys - In `ha_elect_evaluate_and_signal` COMMITTING path: call `push_config_to_standbys(c, ...)` to push config to standbys with lower `config_tag` - Implement `ha_send_ctrlhello_to_all_standbys(c)`: iterate `ha->standby_endpoints`, send `CtrlHelloRequest` to each via `xce_request` — triggers existing catch-up replication in `handle_ctrlhello` - Observable: 3-node startup log shows primary pushing config to ctrl-b/c immediately after COMMITTING; ctrl-b/c logs show config received diff --git a/src/controller/cluster.c b/src/controller/cluster.c index 3de63419..b1c7e23a 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -920,15 +920,25 @@ int cluster_init_phase2(cluster_t* c) vtaskmgr_init(&(c->vtaskmgr), nvcpus); } - delaymgr_init(&(c->delaymgr), 200 /* precision_ms */, run_delay_tasks); + /* 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. */ - { + * 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 diff --git a/src/controller/myapp.c b/src/controller/myapp.c index 9dde2e55..8e342ebb 100644 --- a/src/controller/myapp.c +++ b/src/controller/myapp.c @@ -1,28 +1,58 @@ +#define _GNU_SOURCE #include -#include +#include #include #include +#include #include #include +#include + #include "cluster.h" #include "ha.h" #include "myconf.h" #include "netio.h" -#include "xce/cluster_def.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; -int myapp_run(myconf_t* myconf) +/* + * 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, + * bootstrap_config_if_needed, 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; @@ -30,7 +60,7 @@ int myapp_run(myconf_t* myconf) memset(&c, 0x0, sizeof(c)); c.myconf = myconf; - /* Phase 1: pre-config init (hooks, HA priority, standby_endpoints, cbuf) */ + /* ── Step 1: Phase 1 init ─────────────────────────────────────────────── */ rc = cluster_init_phase1(&c); if (rc < 0) { ERROR("cluster_init_phase1: %s", strerror(-rc)); @@ -38,9 +68,21 @@ int myapp_run(myconf_t* myconf) return rc; } - /* SSL context — created after Phase 1, used by election probing and Phase 2 */ + /* ── 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(); - SSL_CTX* ssl_ctx = create_ssl_context(myconf->certfile, myconf->keyfile); + + /* 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); @@ -48,7 +90,19 @@ int myapp_run(myconf_t* myconf) } c.ha.ssl_ctx = ssl_ctx; - /* Req 1.1: open the listener socket before any election probe so that + /* 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) { @@ -57,9 +111,8 @@ int myapp_run(myconf_t* myconf) return rc; } - /* Req 1.2: initialise the I/O manager and virtual task manager before - * election so that incoming vote requests are handled by the normal - * dispatch loop once it starts. */ + /* 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); @@ -68,17 +121,13 @@ int myapp_run(myconf_t* myconf) vtaskmgr_init(&(c.vtaskmgr), nvcpus); } - /* Req 1.2: start the I/O worker thread so that incoming OP_requestVote RPCs - * from peers are accepted and dispatched during the election phase. - * Without iomgr_start here, connections queue in the kernel (Recv-Q > 0) - * but no thread reads them, causing peers to time out waiting for vote replies. */ iomgr_start(&(c.iomgr)); vtaskmgr_start(&(c.vtaskmgr)); - /* Register the listener socket with epoll NOW so that iomgr can accept - * and dispatch incoming OP_requestVote connections during the election. - * Without this, iomgr runs but has no fd to watch — SSL_connect from - * peers succeeds at TCP level but hangs waiting for SSL_accept. */ + /* ── 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) { @@ -88,12 +137,50 @@ int myapp_run(myconf_t* myconf) } } + /* ── 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. This allows the freshness gate - * in am_i_primary_quorum() (Req 2.4) to work correctly: a node with a valid - * local config will win over nodes that have no config (tag=0). - * If no config file exists yet, config_version_tag stays 0 and the node will - * defer the config fetch to after election via bootstrap_config_if_needed. */ + * 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) { @@ -103,125 +190,40 @@ int myapp_run(myconf_t* myconf) /* Ignore errors — node may have no config yet (first boot or standby) */ } - /* Req 3.1–3.4: probe peers for config tags before election so that a node - * joining an existing cluster knows to defer its config fetch to post-election - * rather than participating as a first-boot candidate with no config. */ - rc = bootstrap_config_if_needed(&c); - if (rc != 0) { - ERROR("bootstrap_config_if_needed: %s", strerror(-rc)); - cluster_fini(&c); - return rc; - } + /* ── Step 8: ha_start_election_async ─────────────────────────────────── */ - /* Election retry loop: keep running until this node becomes PRIMARY or - * confirms a primary exists (via heartbeat or election loss). - * - * Iteration: - * 1. Wait for heartbeat (election_timeout_ms + jitter). - * - Heartbeat received → existing primary detected → STANDBY, exit loop. - * - Timeout → no primary visible, proceed to election. - * 2. Run election (2 rounds with jitter). - * - Won PRIMARY → exit loop as primary. - * - Lost (split vote or peer won) → loop back to step 1. - * The next heartbeat wait will catch the winner's heartbeat if one exists, - * or fire after election_timeout_ms if it was a split vote. - */ - ctrl_role_t final_role = CTRL_ELECTING; - - /* Build peer pointer array once (stable across iterations) */ + /* 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) + if (i != c.ha.my_priority) { peer_ptrs[n_other++] = (xce_candidate_t *)&pcfg->candidates[i]; - } - - for (;;) { - /* Step 1: wait for heartbeat from an existing primary */ - ctrl_role_t hb_role = ha_wait_for_heartbeat_or_timeout(&c.ha); - if (hb_role == CTRL_STANDBY) { - INFO("HA: heartbeat received — existing primary detected, becoming STANDBY"); - final_role = CTRL_STANDBY; - break; - } - - /* Step 2: no heartbeat — run election */ - uint64_t my_tag = myconf->config_version_tag; - uint64_t my_lsn = (uint64_t)atomic_load_explicit(&c.ha.applied_lsn, - memory_order_acquire); - ctrl_role_t elected = am_i_primary_quorum( - &c.ha, n_peers, (n_other > 0) ? peer_ptrs : NULL, - my_tag, my_lsn, myconf->name); - - if (elected == CTRL_PRIMARY) { - final_role = CTRL_PRIMARY; - break; - } - - /* Lost election — loop back; next heartbeat wait will resolve role */ - INFO("HA: election inconclusive — retrying after heartbeat wait"); - } - - /* Post-loop: apply role */ - if (final_role == CTRL_PRIMARY) { - rc = load_primary_config(&c); - if (rc != 0) { - cluster_fini(&c); - myconf_fini(myconf); - return rc; - } - atomic_store_explicit(&c.ha.role, (int32_t)CTRL_PRIMARY, - memory_order_release); - } else { - /* STANDBY: fetch config from primary if not already loaded */ - if (c.ha.defer_config_fetch) { - if (pcfg->candidates_count > 0) - INFO("HA bootstrap: deferred config fetch from winner %s:%u", - pcfg->candidates[0].hostname, (unsigned)pcfg->candidates[0].port); - c.ha.defer_config_fetch = false; - } - if (myconf->config_version_tag == 0) { - rc = fetch_config_from_primary(&c); - if (rc != 0) { - cluster_fini(&c); - myconf_fini(myconf); - return rc; - } - } else { - INFO("Standby has local config (tag=%016llx) — skipping fetch from primary", - (unsigned long long)myconf->config_version_tag); } } - /* Req 4.4: warn if config file contains deprecated 'vcpus' key */ - { - char warn_buf[256] = {0}; - validate_startup_config(myconf->conffile, warn_buf, sizeof(warn_buf)); - if (warn_buf[0]) { - WARN("startup: %s", warn_buf); - } - } + uint64_t my_tag = myconf->config_version_tag; + uint64_t my_lsn = (uint64_t)atomic_load_explicit(&c.ha.applied_lsn, + memory_order_acquire); - /* Phase 2: full init (requires cluster_config) */ - rc = cluster_init_phase2(&c); - if (rc < 0) { - ERROR("cluster_init_phase2: %s", strerror(-rc)); - cluster_fini(&c); - myconf_fini(myconf); - return rc; - } + /* 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); - rc = cluster_start(&c); - if (rc < 0) { - SET_ERROR(-rc); - perror("cluster_start"); - cluster_fini(&c); - myconf_fini(myconf); - return rc; - } + /* ── 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); diff --git a/src/controller/testsuite/test_ha_startup.c b/src/controller/testsuite/test_ha_startup.c index 9bb1da24..4f47dc17 100644 --- a/src/controller/testsuite/test_ha_startup.c +++ b/src/controller/testsuite/test_ha_startup.c @@ -41,6 +41,12 @@ int validate_vcpus(int vcpus); int validate_startup_config(const char *config_path, char *warn_buf, size_t warn_len); +/* Task 5.1 — ha_start_election_async (implemented in ha.c, task 2.4) */ +void ha_start_election_async(ha_state_t *ha, cluster_t *c, + int n_peers, xce_candidate_t **peers, + uint64_t my_config_tag, uint64_t my_lsn, + const char *my_name); + /* ── Dispatch gate (task 3.3) — does NOT exist yet; linker failure expected ── * * ha_dispatch_gate: role-based operation filter for the vtaskmgr dispatch loop. @@ -583,6 +589,105 @@ static void test_dispatch_gate_committing_allows_repl_ops(void) printf("test_dispatch_gate_committing_allows_repl_ops: PASS\n"); } +/* ── Task 5.1: myapp_run startup sequence ──────────────────────────────── + * + * Req 1.1, 9.3, 9.4: myapp_run must call ha_start_election_async (non-blocking) + * and then cluster_run without any intervening blocking calls. + * + * These tests verify the call-sequence contract by checking that: + * 1. ha_start_election_async with n_peers=1 sets ha->n_peers = 1 (single-node + * path — the same path myapp_run takes for a single-node cluster). + * 2. The role after ha_start_election_async(n_peers=1) is CTRL_PRIMARY_COMMITTING + * (not CTRL_PRIMARY) — matching what myapp_run would produce in the new design. + * 3. ha_start_election_async completes in < 10ms — confirming myapp_run does not + * block before entering cluster_run. + * + * These tests are intentionally thin: myapp_run itself is an integration entry + * point that requires a live network socket. The unit tests here verify that + * the building blocks behave correctly; the integration test is test 13 in the + * design testing strategy (regression: single-node self-elects). + */ + +static ha_config_t g_startup_cfg = { + .heartbeat_interval_ms = 3000, + .heartbeat_timeout_ms = 10000, + .election_timeout_ms = 2000, + .probe_connect_timeout_ms = 500, + .max_promotion_retries = 5, + .election_jitter_max_ms = 0, + .election_priority_step_ms = 500, + .committing_timeout_ms = 10000, +}; + +static void test_nonblocking_single_node_election_sets_n_peers(void) +{ + /* Req 1.1, 9.3: the new myapp_run path calls ha_start_election_async with + * n_peers=1 for a single-node cluster. This test verifies the call stores + * n_peers in ha->n_peers, matching the new myapp_run contract. */ + ha_state_t ha; + ha_init(&ha, &g_startup_cfg, 0); + ha.n_peers = 0; + ha.phase2_done = 0; + + ha_start_election_async(&ha, NULL, 1, NULL, 0ULL, 0ULL, "ctrl-a"); + + TEST_ASSERT_EQUAL_INT_MESSAGE(1, ha.n_peers, + "myapp_run path: ha_start_election_async must store n_peers=1 for single-node"); + + ha_fini(&ha); + printf("test_nonblocking_single_node_election_sets_n_peers: PASS\n"); +} + +static void test_nonblocking_single_node_election_enters_committing(void) +{ + /* Req 9.3, 8.1: After ha_start_election_async(n_peers=1) the role must be + * CTRL_PRIMARY_COMMITTING, NOT CTRL_PRIMARY. + * This validates that the new myapp_run path (without blocking for(;;) loop) + * correctly transitions through the COMMITTING state. */ + ha_state_t ha; + ha_init(&ha, &g_startup_cfg, 0); + ha.phase2_done = 0; + atomic_store_explicit(&ha.role, (int32_t)CTRL_ELECTING, memory_order_release); + + ha_start_election_async(&ha, NULL, 1, NULL, 100ULL, 50ULL, "ctrl-a"); + + ctrl_role_t role = (ctrl_role_t)atomic_load_explicit( + &ha.role, memory_order_acquire); + TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY_COMMITTING, (int)role, + "myapp_run path: single-node must enter CTRL_PRIMARY_COMMITTING after election"); + + ha_fini(&ha); + printf("test_nonblocking_single_node_election_enters_committing: PASS (role=%d)\n", + (int)role); +} + +static void test_nonblocking_election_returns_in_under_10ms(void) +{ + /* Req 9.4, 1.1: ha_start_election_async must return in < 5ms (design spec). + * Observable: myapp_run exits into cluster_run within 5ms of startup; + * no blocking calls visible between ha_start_election_async and cluster_run. */ + ha_state_t ha; + ha_init(&ha, &g_startup_cfg, 0); + ha.phase2_done = 0; + + struct timespec t0; + clock_gettime(CLOCK_MONOTONIC, &t0); + + ha_start_election_async(&ha, NULL, 1, NULL, 0ULL, 0ULL, "ctrl-a"); + + struct timespec t1; + clock_gettime(CLOCK_MONOTONIC, &t1); + long elapsed_ms_val = (t1.tv_sec - t0.tv_sec) * 1000L + + (t1.tv_nsec - t0.tv_nsec) / 1000000L; + + TEST_ASSERT_LESS_OR_EQUAL_INT_MESSAGE(5, (int)elapsed_ms_val, + "myapp_run path: ha_start_election_async must return in < 5ms (design spec)"); + + ha_fini(&ha); + printf("test_nonblocking_election_returns_in_under_10ms: PASS (%ldms)\n", + elapsed_ms_val); +} + /* ── Test suite entry point ───────────────────────────────────────────── */ void test_ha_startup(void) @@ -634,4 +739,9 @@ void test_ha_startup(void) /* Task 3.1 — dispatch gate: CTRL_PRIMARY_COMMITTING */ RUN_TEST(test_dispatch_gate_committing_denies_job_ops); RUN_TEST(test_dispatch_gate_committing_allows_repl_ops); + + /* Task 5.1 — myapp_run non-blocking startup sequence (Req 1.1, 9.3, 9.4) */ + RUN_TEST(test_nonblocking_single_node_election_sets_n_peers); + RUN_TEST(test_nonblocking_single_node_election_enters_committing); + RUN_TEST(test_nonblocking_election_returns_in_under_10ms); }