myxce/src/controller/myapp.c

#define _GNU_SOURCE
#include <errno.h>
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>

#include <openssl/ssl.h>

#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);
/* ha-term-safety task 4.3: hash-verified recovery-activate from the latest
 * committed WAL_CONFIG_MARKER (-ENOENT falls back to load_primary_config). */
int config_activate_committed(cluster_t *c);
/* bug-1695: bootstrap "leader already known" decision — implemented in
 * startup_validation.c (not myapp.c) so it links into controller_unit_test,
 * which does not build myapp.c (myapp_run is an untestable integration entry
 * point requiring a live network socket). See that function's doc comment
 * in startup_validation.c for the full contract. */
int myapp_leader_already_known(ha_state_t *ha, const char *my_name,
                               uint64_t term_at_boot);

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();

        /* --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, 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, 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, myconf->wal_segment_bytes);

        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;
    }

    /* bug-1416 round 2: close the recovery_complete gate HERE, at the
     * earliest safe moment, not just around persist_recover()'s own call
     * later (cluster_init_phase2, well after election). This node's own
     * WAL/committed-state materialization is not truly "recovered" until
     * that call runs -- and this process can receive and durably APPLY
     * live-replicated entries (ha_apply_single_entry / ha_apply_term_
     * entry_via_library) before then, e.g. while stuck retrying its own
     * election as a candidate that keeps losing/timing out before
     * discovering a live primary and stepping down (confirmed live: a
     * node stuck ELECTING for tens of seconds while already accepting
     * AppendEntries from a genuine leader). Set BEFORE ha_start_election_
     * async/cluster_run() below can process any network traffic, so no
     * live-replication apply is ever reachable before this gate is
     * closed. persist_recover() clears it (and issues a catch-up kick)
     * once its own replay+verification completes. */
    persist_mark_recovery_pending(&c.persist);

    /* ── 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). */
    {
        /* ha-term-safety task 4.3 (remediation r1, Finding B / R7.10): activate
         * the config referenced by the LATEST COMMITTED WAL_CONFIG_MARKER,
         * hash-VERIFIED against the locally-held bytes, instead of a blind
         * load_primary_config. persist recovery (step 6 above) has already
         * re-derived committed_lsn, so config_committed_marker reads the committed
         * prefix here. Semantics:
         *   - committed marker present + bytes hash-match → activate (verified).
         *   - committed marker present + bytes mismatch/absent → loud serve-fence,
         *     member STAYS ALIVE and retries when correct bytes arrive
         *     out-of-band post-election (NEVER self-terminates).
         *   - NO committed marker (legacy / first boot) → -ENOENT falls back to
         *     the existing load_primary_config blind-load (no false fence). */
        int pre_rc = config_activate_committed(&c);
        if (pre_rc == 0) {
            INFO("Pre-election config activated: tag=%016llx",
                 (unsigned long long)c.myconf->config_version_tag);
        } else if (pre_rc == -EAGAIN) {
            ERROR("Pre-election config serve-FENCED: committed config marker "
                  "references bytes not yet present/matching locally — member "
                  "NON-SERVING under the new config, staying alive and awaiting "
                  "out-of-band config-push (R7.9)");
        }
        /* Ignore other errors — node may have no config yet (first boot/standby);
         * it fetches config via the primary's push post-election (Req 7.3). */

        /* config-upgrade-snapshot task 6.2 (Req 5.2): resume Activation-Fence
         * verification here too, unconditionally, so a crash/restart resumes
         * materialization-verification before serving rather than implicitly
         * treating a fresh process's zero-initialized (unfenced) fence as
         * "already verified". Sequenced AFTER config_activate_committed
         * deliberately: that call is what establishes/repairs which config
         * BYTES are locally active for the committed marker; only once that
         * has run does the separate, sequential question
         * upgrade_materialization_verify asks -- has THIS MEMBER'S OWN
         * checkpoint machinery actually materialized under that marker? --
         * become meaningful to (re-)evaluate. persist_recover() (step 6,
         * above) has already restored c.persist.cfgref_* from the latest
         * loaded checkpoint by this point, so the materialization signal
         * this reads is accurate here, at boot, same as it would be after
         * any later checkpoint. Same "ignore other errors, log loud only on
         * the fenced case" treatment as config_activate_committed above --
         * ha_dispatch_gate and the promotion gate (ha.c) are what actually
         * enforce the fence from here on; this call's only job is to seed
         * the correct fence value at boot instead of leaving a freshly-
         * started process at its unfenced zero-default regardless of
         * whether materialization genuinely completed before the crash. */
        int mat_rc = upgrade_materialization_verify(&c);
        if (mat_rc == -EAGAIN) {
            ERROR("Pre-election upgrade materialization FENCED: this "
                  "member's own upgrade-snapshot materialization for the "
                  "committed config marker is incomplete or hash-mismatched "
                  "— member NON-SERVING and NON-PROMOTABLE, staying alive "
                  "and retrying (config-upgrade-snapshot R1.8/3.2/3.4/5.2)");
        }
        /* Ignore other errors/success silently — mirrors the "ignore other
         * errors" handling above; ha_dispatch_gate / the promotion gate
         * enforce the fence going forward. */
    }

    /* ── 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];
        }
    }

    /* Advertise the candidate freshness for the boot RequestVote payload
     * (ha-commit-durability task 4.2).  TAG axis = the config-epoch fold
     * ha_vote_self_freshness computes (matches the voter side); LSN axis = the
     * BARE durable WAL tip persist_durable_floor (via
     * ha_candidate_advertise_lsn).  A candidate may only advertise a log
     * position it can durably back, so a node booted with a truncated log
     * cannot over-advertise a stale volatile learned-committed value and win
     * (design §1a/§5a).  The bare durable tip is still the genuine voter-floor
     * local-state input on the peer side (peers now fold durable_floor, not
     * applied_lsn), so the most-durable node remains grantable — no livelock
     * (design §4). */
    uint64_t local_tag = myconf->config_version_tag;
    uint64_t local_lsn = (uint64_t)atomic_load_explicit(&c.ha.applied_lsn,
                                                         memory_order_acquire);
    uint64_t my_tag = 0, my_lsn = 0;
    ha_vote_self_freshness(&c.ha, local_tag, local_lsn, &my_tag, NULL);
    my_lsn = ha_candidate_advertise_lsn(&c.persist);

    /* 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);

    /* bug-1695: snapshot ha->raft.term immediately after ha_init_raft() seeds
     * it from the just-restored (ha_restore_term(), inside
     * cluster_init_phase1 above) on-disk term — strictly BEFORE the
     * startup-stagger nap below and before any raft callback (vote grant /
     * AppendEntries UpdateTerm) can fire from iomgr traffic processed during
     * the nap.  myapp_leader_already_known() requires raft.term to have
     * genuinely advanced past this baseline — not merely be nonzero — to
     * treat "a leader is known" as true; see that function's doc comment for
     * the full rationale (a merely-restored, never-since-advanced term was
     * trivially nonzero on every restart after the cluster's first election,
     * permanently skipping every later election attempt). */
    uint64_t term_at_boot = c.ha.raft.term;

    /* 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.
     *
     * bug-1695: the raft-term fallback specifically requires raft.term to
     * have ADVANCED past term_at_boot (captured above, right after
     * ha_init_raft()) — not merely be nonzero.  See
     * myapp_leader_already_known()'s doc comment (startup_validation.c) for
     * the full rationale: a bare `> 0` was trivially true on every restart
     * after the cluster's first election, since ha_restore_term() restores a
     * persisted, never-decreasing term. */
    {
        int leader_known = myapp_leader_already_known(&c.ha, myconf->name,
                                                       term_at_boot);
        if (leader_known) {
            pthread_mutex_lock(&c.ha.lock);
            uint64_t vft_log = c.ha.voted_for_term;
            pthread_mutex_unlock(&c.ha.lock);
            INFO("myapp_run: leader already known after startup delay "
                 "(voted_for=%s, voted_term=%" PRIu64 ", raft_term=%" PRIu64
                 ", term_at_boot=%" PRIu64 ", primary_fd=%d) — skipping own "
                 "election",
                 c.ha.voted_for_name[0] ? c.ha.voted_for_name : "(none)",
                 vft_log, c.ha.raft.term, term_at_boot, 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);

            /* bug-849: a follower that REJOINS an existing cluster skips its own
             * election (a primary already exists), so the election-result path
             * that calls cluster_init_phase2 — which builds resources, runs
             * persist_recover, and (bug-848) advertises the recovered
             * applied_lsn — NEVER runs on this path. The node then comes up on
             * an un-recovered WAL (current_lsn=1, applied_lsn=0), the primary
             * re-streams the whole log over its consistent records, and the
             * committed prefix is lost. Run phase2 HERE too (idempotent via
             * phase2_done, same once-per-process contract as the election
             * callers) so EVERY boot path recovers its own WAL and advertises
             * its real position BEFORE cluster_run starts accepting replication.
             * Recovery is decoupled from the primary's config push: a restarted
             * node already loaded its config from disk pre-election. */
            if (!c.ha.phase2_done) {
                int prc = cluster_init_phase2(&c);
                if (prc != 0) {
                    WARN("myapp_run: cluster_init_phase2 on standby rejoin "
                         "failed (rc=%d) — WAL not recovered; member stays "
                         "alive and will retry (bug-1477: this passive "
                         "skip_election path never sees the election-result "
                         "callers that would otherwise retry it)", prc);
                    ha_schedule_phase2_retry(&c);
                } else {
                    c.ha.phase2_done = 1;
                }
            }

            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;
}