Rongsong Shen
9f4292de29
Adds --max-wal-records (mirrors the existing --wal-sync-ms pattern: main.c argp option -> myconf_t -> persist_config_set_max_wal_records()) so a deployment can force the WAL-record-count checkpoint trigger to fire on a job count a soak can actually reach, instead of the production default of 10000. Adds a new `--force-snapshot` soak mode (run-ha-soak.sh): submits enough jobs to cross the configured threshold, kills a non-primary node, wipes its data dir (a genuine empty rejoin), restarts it, and asserts the cluster reconverges with zero walcompare divergence and zero job loss. Fails loud (not a false PASS) if the restarted node's log shows no actual snapshot install, so the round can't silently report success without exercising the real scenario. INVESTIGATION FINDING (not yet resolved, this commit does not close 5.10a or ha-entity-snapshots 7.1): live-tested with temporary diagnostic logging (since reverted). schedule_checkpoint() correctly fires once wal_record_count reaches the configured threshold (confirmed: "wal_record_count=100 max=100" logged), but the delaymgr- scheduled checkpoint_task_fn never executes afterward -- no checkpoint ever completes, no snapshot file is ever written, in a live multi-node deployment. This is a separate, deeper issue in the delaymgr scheduling path (or its wiring for this specific call shape), not a bug in this commit's CLI plumbing or soak scaffolding -- both are built, tested, and confirmed working up to the point where the async checkpoint should run. Needs its own dedicated investigation (likely src/lib delaymgr, not persist.c) before --force-snapshot can produce a PASS. This also has production implications beyond the test scenario: if confirmed, WAL checkpoints may never fire automatically today, meaning WAL segments are never trimmed via this path in a live deployment. controller_unit_test: test_ha_startup 34/0, test_ha_core_safety 28/0, test_persist_checkpoint 5/0, test_walcompare 17/0 (no regression). Soak --selftest 17/17.
391 lines
18 KiB
C
391 lines
18 KiB
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);
|
|
|
|
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);
|
|
|
|
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). */
|
|
{
|
|
/* 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). */
|
|
}
|
|
|
|
/* ── 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);
|
|
|
|
/* 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);
|
|
|
|
/* 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; falling back to "
|
|
"election-path recovery", prc);
|
|
} 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;
|
|
}
|