Rongsong Shen
be4c1f0721
Wires config_committed_marker to a production consumer (config_activate_committed): verifies pushed bytes' sha256 against the committed marker before activation, fences (loud, non-serving, stay-alive) on mismatch, and activates the latest committed marker on recovery. Enforced via ha_dispatch_gate + promotion gate. Fixes bug-981: a 2nd+ config upgrade could permanently fence a standby on a push-before-commit ordering. Adds an off-iomgr delaymgr retry scheduled from the commit-advance hook (atomic read + non-blocking enqueue only on the ACK thread) that re-activates and clears the stale fence once the new marker commits and bytes verify.
385 lines
18 KiB
C
385 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);
|
|
|
|
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;
|
|
}
|