myxce/src/controller/config_repl.c
Rongsong Shen 102e263cce fix(ha): close permanent catch-up wedge from duplicated replication-target formula (bug-1478)
Two independent call sites duplicated target = max(committed_lsn,
last_written): ha.c's heartbeat/stateReport dispatch (repl_ack_decode_and_
dispatch) and config_repl.c's catchup_derive_target (used by the ACK-driven
catch-up continuation path). Once config-upgrade-snapshot's marker-triggered
checkpoint advances last_written one LSN past an already-committed
committed_lsn (the checkpoint's own consumed, never-replicated LSN slot),
the unconditional max() permanently wedges every standby's catch-up target
at an unreachable value -- the primary re-streams an empty range every
heartbeat forever.

Extracted the shared, pure ha_compute_repl_target_lsn(committed_lsn,
last_written) (ha.h/ha.c), which only falls back to last_written on genuine
committed_lsn==0 bootstrap, and pointed both call sites at it so they can
never disagree again.

Fixing the second call site exposed three pre-existing tests in
test_convergence_verification.c that hardcoded the old buggy formula's
output as their expected target; updated their fixture values (committed_lsn
4->9, matching appended_lsn) to decouple them from the now-fixed edge case,
which already has dedicated pure-function coverage in test_ha_role.c.

GREEN: full controller_unit_test suite = 1040 tests, 0 failures, clean
across multiple runs. Live-verified via run-ha-soak.sh
--config-upgrade-wal-trim (18 jobs, zero job loss, ctrl-1 reconverged 3-of-3
using the preserved floor segment) -- unblocks config-upgrade-snapshot
task 8.9.
2026-07-28 09:53:20 +00:00

3069 lines
124 KiB
C
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#define _GNU_SOURCE
#include "config_repl.h"
#include <dirent.h>
#include <errno.h>
#include <fcntl.h>
#include <limits.h>
#include <netdb.h>
#include <openssl/err.h>
#include <openssl/sha.h>
#include <openssl/ssl.h>
#include <poll.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <time.h>
#include <utime.h>
#include <unistd.h>
#include <inttypes.h>
#include <zlib.h> /* bug-1389 triage: per-chunk crc32 instrumentation */
#include "byteorder.h"
#include "clusterconfig.h"
#include "cluster.h"
#include "controller.capnp.h" /* OPCode_upgradeConfigAck, OPCode_walCatchupEntry */
#include "ha.h"
#include "netio.h" /* endpoint_t, xce_request() */
#include "repl.h" /* repl_wait_majority_ack, repl_on_config_ack */
#include "request.h" /* xce_request, xce_reply, init_reply_by_req */
#include "utils.h"
#include "wobj.h"
#include "xce/cluster_def.h"
#include "wal_types.h" /* wal_entry_hdr_t, WAL_* constants */
/* ── WAL_CONFIG_MARKER codec (ha-term-safety task 4.1) ───────────────
* Fixed, padding-independent serialization of a config-change marker:
* 8-byte big-endian version_tag followed by the raw 32-byte SHA-256. NO
* configuration bytes are ever encoded (R7.2/R7.4) — the bytes travel
* out-of-band on the config-push transport. The explicit byte layout (rather
* than a struct memcpy) keeps the form stable across compiler padding and host
* endianness, since this payload replicates to standbys via AppendEntries. */
int config_marker_encode(uint64_t version_tag, const uint8_t sha256[32],
uint8_t *buf, size_t bufsz, size_t *out_len)
{
if (!sha256 || !buf || !out_len)
return -EINVAL;
if (bufsz < WAL_CONFIG_MARKER_PAYLOAD_LEN)
return -ENOSPC;
/* version_tag: 8 bytes, big-endian (matches config_repl wire convention). */
buf[0] = (uint8_t)(version_tag >> 56);
buf[1] = (uint8_t)(version_tag >> 48);
buf[2] = (uint8_t)(version_tag >> 40);
buf[3] = (uint8_t)(version_tag >> 32);
buf[4] = (uint8_t)(version_tag >> 24);
buf[5] = (uint8_t)(version_tag >> 16);
buf[6] = (uint8_t)(version_tag >> 8);
buf[7] = (uint8_t)(version_tag);
/* sha256: raw 32 bytes — the configuration *reference*, never the bytes. */
memcpy(buf + 8, sha256, 32);
*out_len = WAL_CONFIG_MARKER_PAYLOAD_LEN;
return 0;
}
int config_marker_decode(const uint8_t *buf, size_t len,
uint64_t *version_tag, uint8_t sha256_out[32])
{
if (!buf || len < WAL_CONFIG_MARKER_PAYLOAD_LEN)
return -EINVAL;
if (version_tag) {
*version_tag = ((uint64_t)buf[0] << 56) | ((uint64_t)buf[1] << 48) |
((uint64_t)buf[2] << 40) | ((uint64_t)buf[3] << 32) |
((uint64_t)buf[4] << 24) | ((uint64_t)buf[5] << 16) |
((uint64_t)buf[6] << 8) | ((uint64_t)buf[7]);
}
if (sha256_out)
memcpy(sha256_out, buf + 8, 32);
return 0;
}
/* ── Config marker append + committed-config derive (task 4.2) ───────────────
*
* A configuration change is ordered + committed through the SAME (term,lsn) log
* as job state. config_append_marker routes the marker through the single
* primary append entry point (ha_primary_log_append → raft_log_append), so the
* library stamps the live term and assigns the LSN, fires on_persist_entry, and
* the marker inherits the same commit rule and (term,lsn) reconcile/truncate as
* a job entry. The committed configuration is DERIVED from the latest committed
* marker within the committed prefix (config_committed_marker) — there is no
* separate authoritative config commit index. */
int config_append_marker(cluster_t *c, uint64_t version_tag,
const uint8_t sha256[32], uint64_t *out_lsn)
{
if (!c || !sha256)
return -EINVAL;
uint8_t payload[WAL_CONFIG_MARKER_PAYLOAD_LEN];
size_t plen = 0;
int rc = config_marker_encode(version_tag, sha256, payload,
sizeof(payload), &plen);
if (rc != 0)
return rc;
/* Single primary append entry point: stamps term (R7.3), assigns LSN,
* commits through the same barrier (R7.11), reconciles by (term,lsn) (R7.6).
* The config BYTES never enter this payload (R7.2/R7.4). */
return ha_primary_log_append(c, WAL_CONFIG_MARKER, payload,
(uint32_t)plen, out_lsn);
}
/* scan_latest_committed_marker: walk the local WAL segment files and find the
* HIGHEST-LSN WAL_CONFIG_MARKER whose LSN is <= committed_lsn (the committed
* prefix). On a match, decode its {version_tag, sha256} into the out params.
*
* Returns 0 on a found+decoded marker; -ENOENT when no committed marker exists;
* -EIO on an unreadable WAL directory. An uncommitted marker (LSN >
* committed_lsn) is deliberately skipped so a config change is not the committed
* config until its marker reaches the committed prefix (R7.7). */
static int scan_latest_committed_marker(const char *wal_path,
uint64_t committed_lsn,
uint64_t *version_tag,
uint8_t sha256_out[32])
{
if (committed_lsn == 0)
return -ENOENT; /* nothing committed → no committed config marker */
DIR *d = opendir(wal_path);
if (!d)
return -EIO;
/* Enumerate segment start-LSNs ascending so we replay in LSN order and the
* last marker we see at LSN <= committed_lsn is the latest committed one. */
uint64_t seg_lsns[4096];
int nseg = 0;
struct dirent *de;
while ((de = readdir(d)) != NULL &&
nseg < (int)(sizeof(seg_lsns) / sizeof(*seg_lsns))) {
uint64_t sl = 0;
if (sscanf(de->d_name, "seg-%016" SCNx64 ".wal", &sl) == 1)
seg_lsns[nseg++] = sl;
}
closedir(d);
if (nseg == 0)
return -ENOENT;
/* insertion sort ascending */
for (int i = 1; i < nseg; i++) {
uint64_t key = seg_lsns[i];
int j = i - 1;
while (j >= 0 && seg_lsns[j] > key) { seg_lsns[j + 1] = seg_lsns[j]; j--; }
seg_lsns[j + 1] = key;
}
int found = 0;
uint64_t found_tag = 0;
uint8_t found_sha[32];
for (int si = 0; si < nseg; si++) {
char seg_path[4096];
snprintf(seg_path, sizeof(seg_path),
"%s/seg-%016" PRIx64 ".wal", wal_path, seg_lsns[si]);
int fd = open(seg_path, O_RDONLY);
if (fd < 0)
continue;
uint32_t off = 0;
for (;;) {
wal_entry_hdr_t hdr;
ssize_t n = pread(fd, &hdr, sizeof(hdr), (off_t)off);
if (n < (ssize_t)sizeof(hdr)) break;
if (hdr.lsn == 0 && hdr.type == 0 && hdr.payload_len == 0) break;
if (hdr.lsn > committed_lsn)
break; /* past the committed prefix — stop scanning this seg */
if (hdr.type == WAL_CONFIG_MARKER &&
hdr.payload_len >= WAL_CONFIG_MARKER_PAYLOAD_LEN) {
uint8_t buf[WAL_CONFIG_MARKER_PAYLOAD_LEN];
ssize_t pr = pread(fd, buf, sizeof(buf),
(off_t)(off + sizeof(hdr)));
if (pr == (ssize_t)sizeof(buf)) {
uint64_t tag = 0;
if (config_marker_decode(buf, sizeof(buf), &tag,
found_sha) == 0) {
found_tag = tag;
found = 1; /* later markers overwrite — latest wins */
}
}
}
off += (uint32_t)(sizeof(hdr) + hdr.payload_len);
}
close(fd);
}
if (!found)
return -ENOENT;
if (version_tag)
*version_tag = found_tag;
if (sha256_out)
memcpy(sha256_out, found_sha, 32);
return 0;
}
int config_committed_marker(const cluster_t *c, uint64_t *version_tag,
uint8_t sha256_out[32])
{
if (!c)
return -EINVAL;
uint64_t committed = atomic_load_explicit(&c->persist.committed_lsn,
memory_order_acquire);
int rc = scan_latest_committed_marker(c->persist.wal_path, committed,
version_tag, sha256_out);
if (rc == 0 || rc == -EINVAL)
return rc;
/* Compaction fallback (task 4.2 remediation r1): a checkpoint can trim the
* WAL segment holding the latest committed marker, so the live scan above
* returns -ENOENT even though a committed config exists. The checkpoint
* stamped that marker's {version_tag, sha256} reference into the snapshot
* sentinel (persist.cfgref_*, read back at recovery), so fall back to it.
* This keeps the derived committed config stable across compaction and lets
* recovery activate the latest committed marker (R7.7/R7.10) even when no
* marker survives in the live WAL. */
if (c->persist.cfgref_present) {
if (version_tag)
*version_tag = c->persist.cfgref_tag;
if (sha256_out)
memcpy(sha256_out, c->persist.cfgref_sha, 32);
return 0;
}
return rc; /* genuinely no committed config marker anywhere */
}
/* ── Config byte delivery: hash-verify-before-activate + serve-fence (task 4.3) ─
*
* config_committed_marker (4.2) is the AUTHORITY for which configuration is in
* effect; the configuration BYTES travel out-of-band on the existing config-push
* transport (OPCode_replConfigPush / handle_repl_config_push), never inside a WAL
* payload (R7.4/R7.5). config_activate_committed keys that byte delivery to the
* committed marker's {version_tag, sha256}: it VERIFIES the locally-held bytes'
* sha256 against the committed marker BEFORE activating (R7.8), and on a mismatch
* (or bytes absent locally) keeps the member from serving under the new config and
* KEEPS RETRYING rather than activating mismatched config (R7.9) — the loud
* non-serving fence that NEVER self-terminates. On recovery it activates the config
* referenced by the LATEST committed marker (R7.10).
*/
int config_serve_fenced(const cluster_t *c)
{
if (!c)
return 0;
return atomic_load_explicit(&c->ha.config_serve_fence,
memory_order_acquire) ? 1 : 0;
}
/* Read the locally-held config bytes (resolved via myconf->conffile or
* conf/current) and compute their sha256 into out_sha. Returns 0 with the hash on
* success; -ENOENT when no local config bytes exist yet (the out-of-band push has
* not delivered them). The bytes are the out-of-band-delivered config, NEVER a WAL
* payload (R7.4/R7.5). */
static int local_config_sha256(cluster_t *c, uint8_t out_sha[32])
{
myconf_t *mc = c->myconf;
if (!mc || !mc->workdir || !mc->name)
return -EINVAL;
/* Resolve the current locally-held versioned config file. */
char conf_dir[4096];
snprintf(conf_dir, sizeof(conf_dir), "%s/%s", mc->workdir, mc->name);
char fname[256] = {0};
int rc = conf_read_current_filename(conf_dir, fname, sizeof(fname));
if (rc != 0)
return -ENOENT; /* no local bytes yet — caller fences + retries */
char path[4352];
snprintf(path, sizeof(path), "%s/conf/%s", conf_dir, fname);
uint8_t *bytes = NULL;
size_t blen = 0;
rc = conf_read_raw(path, &bytes, &blen);
if (rc != 0)
return -ENOENT;
SHA256(bytes, blen, out_sha);
free(bytes);
return 0;
}
int config_activate_committed(cluster_t *c)
{
if (!c)
return -EINVAL;
/* 1. The committed marker is the authority for the target config (R7.7/R7.10:
* on recovery this is the LATEST committed marker). */
uint64_t marker_tag = 0;
uint8_t marker_sha[32];
int rc = config_committed_marker(c, &marker_tag, marker_sha);
if (rc == -ENOENT) {
/* No committed config marker exists (legacy / first boot, or the marker
* has not committed yet). Absent-marker is NOT a hash MISMATCH, so it must
* NOT fence the node (a benign "awaiting bytes/marker" state, never a hard
* divergence fence — R7.9/R7.10). FALL BACK to the existing
* load_primary_config behavior so any locally-held bytes are activated
* (config_version_tag is set for the freshness gate and the push ACK).
* A missing local config file (-ENOENT from load_primary_config) is also
* benign first-boot — the node fetches config post-election. Clear any
* stale fence either way: an absent marker is never a divergence. */
atomic_store_explicit(&c->ha.config_serve_fence, 0, memory_order_release);
int lrc = load_primary_config(c);
if (lrc != 0 && lrc != -ENOENT)
WARN("config_activate_committed: no committed marker; "
"load_primary_config fallback rc=%d", lrc);
return 0;
}
if (rc != 0)
return rc; /* -EINVAL / -EIO surfaced to the caller */
/* 2. Hash-verify the LOCALLY-HELD (out-of-band-delivered) bytes against the
* committed marker BEFORE activating (R7.8). */
uint8_t local_sha[32];
rc = local_config_sha256(c, local_sha);
bool verified = (rc == 0 && memcmp(local_sha, marker_sha, 32) == 0);
if (!verified) {
/* 3. MISMATCH or bytes not yet present locally: the bytes for this
* committed marker have not arrived/verified out-of-band. Keep the
* member NON-SERVING under the new config and KEEP RETRYING — the loud
* non-self-terminating fence (R7.9). NEVER activate mismatched bytes.
* Stays ALIVE: returns -EAGAIN so the caller retries when the matching
* bytes arrive via the existing config-push transport. */
int was = atomic_exchange_explicit(&c->ha.config_serve_fence, 1,
memory_order_acq_rel);
if (!was)
ERROR("config serve-fence: committed config marker tag=%016" PRIx64
" references bytes whose sha256 is not present/matching locally "
"(rc=%d) -- member NON-SERVING under the new config, awaiting "
"out-of-band config-push bytes; staying alive and retrying",
marker_tag, rc);
return -EAGAIN;
}
/* 4. VERIFIED: the locally-held bytes hash-match the committed marker —
* activate the configuration (load it into myconf), then clear the fence. */
rc = load_primary_config(c);
if (rc != 0) {
/* The verified bytes failed to load (e.g. decode error). Stay fenced and
* retrying rather than serving an unloadable config — never self-terminate. */
int was = atomic_exchange_explicit(&c->ha.config_serve_fence, 1,
memory_order_acq_rel);
if (!was)
ERROR("config serve-fence: hash-verified config bytes for committed "
"marker tag=%016" PRIx64 " failed to load (rc=%d) -- member "
"NON-SERVING, staying alive and retrying", marker_tag, rc);
return rc;
}
atomic_store_explicit(&c->ha.config_serve_fence, 0, memory_order_release);
INFO("config activated from committed marker tag=%016" PRIx64
" (hash-verified out-of-band bytes)", marker_tag);
return 0;
}
/* ── Commit-advance retry trigger (task 4.3 remediation r2, Finding D / R7.9) ──
*
* The config MARKER (WAL stream) and the config BYTES (out-of-band config-push)
* arrive on INDEPENDENT transports. On a 2nd-or-later upgrade the bytes can be
* saved locally BEFORE the standby's committed prefix advances past the NEW
* marker: at that moment config_committed_marker still returns the OLDER marker,
* so the freshly-saved NEW bytes hash-MISMATCH it and config_activate_committed
* sets the serve-fence (-EAGAIN). The NEW marker commits later via the WAL path,
* but on_commit_advanced runs on the iomgr ACK thread and MUST NOT do disk I/O,
* so it cannot re-activate inline; and the primary's "config-behind" re-push net
* does NOT fire because conf_save_config already advanced this member's advertised
* config_version_tag. Without a retry the standby stays PERMANENTLY fenced (and
* non-promotable) for that upgrade until reboot — a §7.9 liveness violation.
*
* The fix: when commit advances WHILE this member is serve-fenced, schedule a
* one-shot task that re-runs config_activate_committed off the iomgr ACK thread.
* The delaymgr thread itself must never block (ha-async-io.md thread-role
* contract — it also drives heartbeats and every election timeout), so the
* one-shot delaymgr task is only a trigger: it hands the actual blocking disk
* reads (scan_latest_committed_marker pread + load_primary_config) to a
* VTASK_LIVE_BLOCK vtask and returns immediately (2026-07-09 correction — this
* comment previously and incorrectly claimed the delaymgr worker itself was
* "allowed to block on the disk reads"; that predates the corrected contract
* ha-async-io.md documents). Once the NEW marker is in the committed prefix AND
* the locally-held bytes hash-match it, the retry activates and CLEARS the
* fence; a genuine still-MISMATCH (or no-bytes-yet) keeps the fence and the next
* commit-advance reschedules. -ENOENT still falls back without fencing (the
* no-marker path is benign). Idempotent + self-limiting: the scheduler is a
* no-op once unfenced, so there is no busy-spin. */
/* config_activate_retry_fn: re-runs config_activate_committed so a freshly-
* committed config marker is activated — and a stale serve-fence cleared —
* once its out-of-band bytes verify. Does the actual blocking disk I/O; callers
* must invoke it off the delaymgr thread (see config_activate_retry_vtask_entry
* / config_schedule_activate_retry below). Public so tests can drive it
* directly without a vtask pool. */
int config_activate_retry_fn(void *arg)
{
cluster_t *c = (cluster_t *)arg;
if (!c)
return -EINVAL;
if (!config_serve_fenced(c))
return 0; /* already cleared by another path — nothing to retry */
int rc = config_activate_committed(c);
if (rc == 0)
INFO("config commit-advance retry: serve-fence CLEARED — newly committed "
"config marker activated (hash-verified out-of-band bytes)");
else if (rc == -EAGAIN)
TRACE("config commit-advance retry: still fenced (marker/bytes not yet "
"matching) — will retry on the next commit advance");
else
WARN("config commit-advance retry: config_activate_committed rc=%d", rc);
return rc;
}
/* config_activate_retry_vtask_entry: vtask entry wrapping config_activate_
* retry_fn so its blocking disk I/O runs on a vtask worker, never inline on
* the single delaymgr thread. */
static int config_activate_retry_vtask_entry(vtask_t *task)
{
cluster_t *c = (cluster_t *)task->data->u.generic;
config_activate_retry_fn(c);
return 0;
}
/* config_activate_retry_delaymgr_trampoline: the delaymgr task body. Delaymgr
* must only trigger + enqueue, never block — dispatch the actual retry work to
* a vtask (VTASK_LIVE_BLOCK, the same class the vote-grant term fsync uses,
* R4.6) and return immediately. Fails loud and runs inline (stay alive, never
* silently drop a scheduled retry) if the pool is unavailable. */
static int config_activate_retry_delaymgr_trampoline(void *arg)
{
cluster_t *c = (cluster_t *)arg;
if (!c)
return -EINVAL;
vtaskmgr_t *pool = &c->vtaskmgr;
vtask_data_t *data = calloc(1, sizeof(*data));
vtask_t *task = vtask_alloc();
if (data == NULL || task == NULL) {
WARN("config commit-advance retry: vtask alloc failed — running "
"inline (stay alive)");
free(data);
free(task);
return config_activate_retry_fn(c);
}
data->vtype = VTASK_GENERIC;
data->u.generic = c;
vtask_init(task, (vtask_func_t)config_activate_retry_vtask_entry, data);
if (vtask_enq_classified(pool, task, VTASK_LIVE_BLOCK) < 0) {
WARN("config commit-advance retry: vtask enqueue failed — running "
"inline (stay alive)");
free(data);
free(task);
return config_activate_retry_fn(c);
}
return 0;
}
/* config_schedule_activate_retry: enqueue the retry trampoline on the delaymgr
* (NON-BLOCKING — delay_task_ms only does ordq_put; the trampoline itself
* offloads the blocking work to a vtask, see above). Called from the
* commit-advance path (iomgr ACK thread) so activation's disk I/O runs OFF that
* thread (ha-async-io invariant). No-op when the member is NOT serve-fenced (so
* the steady state schedules nothing) or when the delaymgr is not running (the
* unit-test fixture path, which never inits delaymgr — avoids leaking a task into
* an un-drained queue). Returns 0 on enqueue/no-op, negative on enqueue failure. */
int config_schedule_activate_retry(cluster_t *c)
{
if (!c)
return -EINVAL;
if (!config_serve_fenced(c))
return 0; /* steady state: nothing to retry, schedule nothing */
if (!c->delaymgr.running)
return 0; /* delaymgr not running (test fixture) — caller drives directly */
/* One-shot, minimal delay: run promptly but off the iomgr ACK thread. */
int rc = delay_task_ms(&c->delaymgr, 1, 1,
(delay_task_entry_fn_t)config_activate_retry_delaymgr_trampoline,
c);
if (rc < 0)
WARN("config commit-advance retry: delay_task_ms failed: %d — fence will "
"clear on a later commit advance or config-push", rc);
return rc;
}
/* config_commit_change (ha-term-safety task 4.2 — LIVE wiring): order a config
* change through the unified (term,lsn) log as the CONSENSUS RECORD and make the
* derived committed marker authoritative, retiring committed_config_tag to a
* non-authoritative cache.
*
* 1. Append a WAL_CONFIG_MARKER ({version_tag, sha256} — NO bytes, R7.2/R7.4)
* via config_append_marker → ha_primary_log_append → raft_log_append. The
* library stamps the live term (R7.3), assigns the LSN, and the marker
* inherits the same (term,lsn) reconcile/truncate as a job entry (R7.6).
* This is the ORDERING authority: config_committed_marker derives the
* committed config from it (R7.7).
* 2. Deliver the config BYTES out-of-band and wait for the existing
* majority-ack barrier via push_config_to_standbys (R7.5/R7.11 — byte
* delivery/verify/activate stays task 4.3; the barrier MECHANISM is
* unchanged). push_config_to_standbys advances committed_config_tag as the
* CACHE on success; consumers read the derived marker as the authority.
*
* The marker append precedes the barrier so the consensus record is in the log
* before the change is acknowledged committed. A marker-append failure is fatal
* (fail-loud): the upgrade cannot be ordered, so it returns the append rc and
* the byte barrier is not attempted.
*
* Returns 0 on a committed config change; the barrier rc (-ETIMEDOUT/-EINVAL)
* when the change never reaches majority; a negative append rc on marker-append
* failure. *acked_out (may be NULL) receives the out-of-band transport push
* count (unchanged from push_config_to_standbys' acked semantics). */
int config_commit_change(cluster_t *c, const uint8_t *bytes, size_t blen,
const uint8_t sha256[32], uint64_t version_tag,
int *acked_out)
{
if (acked_out)
*acked_out = 0;
if (!c || !sha256)
return -EINVAL;
/* 1. Append the consensus record: the WAL_CONFIG_MARKER (ordering authority
* for the derived committed config). Fail-loud on append failure — the
* change cannot be ordered, so it is not committed. */
uint64_t marker_lsn = 0;
int rc = config_append_marker(c, version_tag, sha256, &marker_lsn);
if (rc != 0) {
ERROR("config_commit_change: marker append failed (rc=%d, tag=%016"
PRIx64 ") — config change not ordered", rc, version_tag);
return rc;
}
/* 2. Out-of-band byte delivery + the existing majority-ack barrier (the
* commit barrier mechanism is unchanged). On success it advances the
* committed_config_tag CACHE; the derived marker (config_committed_marker)
* remains the authority consumers read. */
rc = push_config_to_standbys(c, bytes, blen, sha256, version_tag, acked_out);
if (rc != 0) {
WARN("config_commit_change: config change did not reach majority"
" (rc=%d, marker_lsn=%" PRIu64 ", tag=%016" PRIx64 ")",
rc, marker_lsn, version_tag);
return rc;
}
INFO("Config change committed (marker_lsn=%" PRIu64 ", tag=%016" PRIx64 ")",
marker_lsn, version_tag);
return 0;
}
/* SSL context for xce_probe_config_tag.
* Stored as void* to avoid SSL_CTX type at this point in the translation unit. */
static void *s_probe_ssl_ctx = NULL;
/* ── config load helpers (shared with myconf.c logic) ─────── */
#define SET_DEFAULT_IF_UNSET(var, unset, val) \
do { \
if ((var) == (unset)) { \
(var) = (val); \
} \
} while (0)
static void fill_default_cluster_config(cluster_config_t *c)
{
SET_DEFAULT_IF_UNSET(c->heartbeat_interval, 0, 5);
SET_DEFAULT_IF_UNSET(c->fetch_job_interval, 0, 10);
}
static int load_cluster_config_from_file(char *file, cluster_config_t **config)
{
struct capn c;
ClusterConfig_ptr p;
FILE *fp;
int rc = 0;
fp = fopen(file, "rb");
if (fp == NULL) {
rc = -errno;
return rc;
}
capn_init_fp(&c, fp, 0);
p.p = capn_getp(capn_root(&c), 0, 1);
decode_ClusterConfig_ptr(config, p);
capn_free(&c);
fclose(fp);
return 0;
}
/* ── load_primary_config ────────────────────────── */
#ifdef XCE_TESTING
/* load_primary_config_race_hook: test-only synchronization seam (same
* pattern as push_thread_send_hook / xce_wal_stream_request_hook). When
* non-NULL, load_primary_config() calls this immediately after capturing the
* OLD myconf->cluster_config pointer this invocation is about to retire
* (free), right before it actually retires it -- letting a unit test
* deterministically force a SECOND, real, concurrent load_primary_config
* call through this exact window instead of depending on scheduler timing.
* NULL in production -- no behaviour change. See
* test_bug1476_concurrent_load_primary_config_no_uaf_or_double_free
* (test_config_repl.c) for the RED-phase reproduction this was built for:
* with the lock below removed, this hook deterministically reproduces the
* live ASan heap-use-after-free bug-1476 caught (two concurrent
* load_primary_config calls racing the free of the SAME myconf->
* cluster_config pointer). With the lock, a second invocation can never
* even reach this hook while the first is parked here -- it blocks
* acquiring myconf->lock before getting this far, which is the
* serialization proof the fix requires. */
void (*load_primary_config_race_hook)(void) = NULL;
#endif
int load_primary_config(cluster_t *c)
{
myconf_t *myconf = c->myconf;
/* bug-1476: serialize the whole free-decode-publish sequence against a
* concurrent load_primary_config() call for the SAME myconf_t. Held
* across every return path below, including the success path — a
* second caller must block here before it can even reach the race hook
* fired near the retire step, not merely around that step alone. */
mylock_lock(&myconf->lock);
/* Build conf dir path: <workdir>/<name> */
char conf_dir[4096];
snprintf(conf_dir, sizeof(conf_dir), "%s/%s",
myconf->workdir, myconf->name);
/* Resolve conf/current → versioned filename */
char fname[256] = {0};
int rc = conf_read_current_filename(conf_dir, fname, sizeof(fname));
if (rc != 0) {
INFO("No config at '%s/conf/current' (rc=%d) — "
"will fetch from primary after election",
conf_dir, rc);
mylock_unlock(&myconf->lock);
return -ENOENT;
}
/* Heap-allocate conffile so it survives past this stack frame */
free(myconf->conffile);
myconf->conffile = NULL;
if (asprintf(&myconf->conffile, "%s/conf/%s", conf_dir, fname) < 0) {
mylock_unlock(&myconf->lock);
return -ENOMEM;
}
/* Read raw bytes and check size */
uint8_t *bytes = NULL;
size_t blen = 0;
rc = conf_read_raw(myconf->conffile, &bytes, &blen);
if (rc != 0) {
ERROR("Primary config '%s' unreadable: %d", myconf->conffile, rc);
free(myconf->conffile);
myconf->conffile = NULL;
mylock_unlock(&myconf->lock);
return rc;
}
/* Compute and cache SHA-256 */
SHA256(bytes, blen, myconf->config_sha256);
uint64_t tag;
memcpy(&tag, myconf->config_sha256, 8);
myconf->config_version_tag = be64toh(tag);
/* Decode Cap'n Proto into a LOCAL temporary first (bug-1476 fix part 2):
* decode_ClusterConfig_ptr calloc's and publishes its out-parameter
* pointer BEFORE decode_ClusterConfig populates the nested fields, so
* decoding directly into the shared myconf->cluster_config field would
* expose any other reader of that field to a live-but-incomplete
* object. Only a fully decoded and validated config is ever published
* below. */
cluster_config_t *new_config = NULL;
rc = load_cluster_config_from_file(myconf->conffile, &new_config);
if (rc != 0) {
ERROR("Primary config '%s' failed Cap'n Proto decode: %d",
myconf->conffile, rc);
free(bytes);
mylock_unlock(&myconf->lock);
return rc;
}
if (new_config == NULL) {
ERROR("Primary config '%s' decoded to NULL", myconf->conffile);
free(bytes);
mylock_unlock(&myconf->lock);
return -EINVAL;
}
free(bytes);
fill_default_cluster_config(new_config);
/* Check for reserved lost_and_found name */
if (config_contains_reserved_name(new_config)) {
ERROR("Config '%s' contains reserved name 'lost_and_found'",
myconf->conffile);
free_ClusterConfig(new_config);
free(new_config);
mylock_unlock(&myconf->lock);
return -EINVAL;
}
/* Default generation to 1 if not set */
if (new_config->config_generation == 0)
new_config->config_generation = 1;
myconf->config_generation = new_config->config_generation;
/* Publish: retire the prior config only now that the new one is fully
* decoded and validated — myconf->cluster_config never points at a
* freed or partially-built object. */
if (myconf->cluster_config) {
#ifdef XCE_TESTING
if (load_primary_config_race_hook)
load_primary_config_race_hook();
#endif
free_ClusterConfig(myconf->cluster_config);
free(myconf->cluster_config);
}
myconf->cluster_config = new_config;
/* Check staleness */
myconf->conf_stale = conf_check_stale(conf_dir, myconf->conf_stale_timeout);
if (myconf->conf_stale)
WARN("Primary config appears stale (age > %lds)",
(long)myconf->conf_stale_timeout);
INFO("Primary config loaded: tag=%016" PRIx64 " gen=%" PRIu64,
myconf->config_version_tag, myconf->config_generation);
mylock_unlock(&myconf->lock);
return 0;
}
/* ── internal helpers ──────────────────────────── */
static void make_conf_dir(const char *workdir, char *out, size_t sz)
{
snprintf(out, sz, "%s/conf", workdir);
}
static void bytes_to_hex(const uint8_t *b, size_t n, char *out)
{
static const char hex[] = "0123456789abcdef";
for (size_t i = 0; i < n; i++) {
out[2 * i] = hex[(b[i] >> 4) & 0xf];
out[2 * i + 1] = hex[b[i] & 0xf];
}
out[2 * n] = '\0';
}
/* ── conf_read_current_filename ─────────────────── */
int conf_read_current_filename(const char *workdir, char *out, size_t outsz)
{
char head_path[4096];
snprintf(head_path, sizeof(head_path), "%s/conf/current", workdir);
FILE *f = fopen(head_path, "r");
if (!f)
return -ENOENT;
struct stat st;
if (fstat(fileno(f), &st) == 0 && st.st_size == 0) {
fclose(f);
ERROR("conf/current is empty — treating as missing config (pointer file"
" may be from an interrupted write)");
return -ENODATA;
}
if (!fgets(out, (int)outsz, f)) {
fclose(f);
return -EIO;
}
fclose(f);
out[strcspn(out, "\n")] = '\0';
if (strchr(out, '/')) {
ERROR("conf/current contains path separator — corrupted pointer file");
return -EINVAL;
}
if (strncmp(out, "cluster.bin.", 12) != 0) {
ERROR("conf/current '%s' does not start with 'cluster.bin.'", out);
return -EINVAL;
}
const char *hex = out + 12;
if (strlen(hex) != 12 || strspn(hex, "0123456789abcdef") != 12) {
ERROR("conf/current suffix '%s' is not 12 lowercase hex chars", hex);
return -EINVAL;
}
return 0;
}
/* ── conf_save_config ───────────────────────────── */
int conf_save_config(myconf_t *myconf, const uint8_t *bytes, size_t len,
const uint8_t sha256[32])
{
/* effective_dir = workdir/name — must match load_primary_config() path */
char effective_dir[4096];
snprintf(effective_dir, sizeof(effective_dir), "%s/%s",
myconf->workdir, myconf->name);
char conf_dir[4096];
make_conf_dir(effective_dir, conf_dir, sizeof(conf_dir));
if (mkdir(conf_dir, 0700) != 0 && errno != EEXIST)
return -errno;
/* Derive 12-char hex tag from first 6 bytes of SHA-256 */
char hex_tag[13];
bytes_to_hex(sha256, 6, hex_tag);
/* Step 1: write versioned file via thread-unique .tmp */
char tmp_path[4096], ver_path[4096];
snprintf(tmp_path, sizeof(tmp_path), "%s/cluster.bin.%s.%lu.tmp",
conf_dir, hex_tag, (unsigned long)pthread_self());
snprintf(ver_path, sizeof(ver_path), "%s/cluster.bin.%s",
conf_dir, hex_tag);
int fd = open(tmp_path, O_WRONLY | O_CREAT | O_TRUNC, 0600);
if (fd < 0)
return -errno;
if (write(fd, bytes, len) != (ssize_t)len) {
close(fd);
unlink(tmp_path);
return -EIO;
}
if (fsync(fd) != 0) {
close(fd);
unlink(tmp_path);
return -EIO;
}
close(fd);
if (rename(tmp_path, ver_path) != 0) {
unlink(tmp_path);
return -errno;
}
/* Step 2: update HEAD pointer via thread-unique .tmp */
char ptr_tmp[4096], ptr_path[4096];
snprintf(ptr_tmp, sizeof(ptr_tmp), "%s/current.%lu.tmp",
conf_dir, (unsigned long)pthread_self());
snprintf(ptr_path, sizeof(ptr_path), "%s/current", conf_dir);
char target[NAME_MAX];
snprintf(target, sizeof(target), "cluster.bin.%s\n", hex_tag);
fd = open(ptr_tmp, O_WRONLY | O_CREAT | O_TRUNC, 0600);
if (fd < 0)
return -errno;
if (write(fd, target, strlen(target)) != (ssize_t)strlen(target)) {
close(fd);
unlink(ptr_tmp);
return -EIO;
}
if (fsync(fd) != 0) {
close(fd);
unlink(ptr_tmp);
return -EIO;
}
close(fd);
if (rename(ptr_tmp, ptr_path) != 0) {
unlink(ptr_tmp);
return -errno;
}
/* Step 2.5: fsync conf/ directory to persist the rename(2)s */
fd = open(conf_dir, O_RDONLY);
if (fd >= 0) {
fsync(fd);
close(fd);
}
/* Step 3: trim old versions */
conf_trim_old_versions(effective_dir, myconf->conf_keep_n);
/* Update conffile to the new versioned path (heap-allocated) */
free(myconf->conffile);
if (asprintf(&myconf->conffile, "%s/cluster.bin.%s", conf_dir, hex_tag) < 0)
return -ENOMEM;
/* Update tag (first 8 bytes of sha256, host order) and full hash */
uint64_t new_tag;
memcpy(&new_tag, sha256, 8);
myconf->config_version_tag = be64toh(new_tag);
memcpy(myconf->config_sha256, sha256, 32);
return 0;
}
/* ── conf_read_raw ──────────────────────────────── */
int conf_read_raw(const char *conffile, uint8_t **bytes, size_t *len)
{
FILE *f = fopen(conffile, "rb");
if (!f)
return -errno;
if (fseek(f, 0, SEEK_END) != 0) {
fclose(f);
return -EIO;
}
long sz = ftell(f);
if (sz < 0) {
fclose(f);
return -EIO;
}
if (fseek(f, 0, SEEK_SET) != 0) {
fclose(f);
return -EIO;
}
if (sz == 0) {
fclose(f);
ERROR("Config file '%s' is empty — not a valid ClusterConfig blob",
conffile);
return -EIO;
}
if ((size_t)sz > CONF_MAX_SIZE) {
fclose(f);
ERROR("Config file '%s' exceeds CONF_MAX_SIZE (%zu bytes)",
conffile, (size_t)sz);
return -EFBIG;
}
*bytes = (uint8_t *)malloc((size_t)sz);
if (!*bytes) {
fclose(f);
return -ENOMEM;
}
if (fread(*bytes, 1, (size_t)sz, f) != (size_t)sz) {
free(*bytes);
*bytes = NULL;
fclose(f);
return -EIO;
}
fclose(f);
*len = (size_t)sz;
return 0;
}
/* ── conf_get_saved_tag ─────────────────────────── */
uint64_t conf_get_saved_tag(const char *workdir)
{
char fname[256] = {0};
if (conf_read_current_filename(workdir, fname, sizeof(fname)) != 0)
return 0;
char ver_path[4096];
snprintf(ver_path, sizeof(ver_path), "%s/conf/%s", workdir, fname);
uint8_t *bytes = NULL;
size_t blen = 0;
if (conf_read_raw(ver_path, &bytes, &blen) != 0)
return 0;
uint8_t sha256[32];
SHA256(bytes, blen, sha256);
free(bytes);
uint64_t tag;
memcpy(&tag, sha256, 8);
return be64toh(tag);
}
/* ── conf_check_stale ───────────────────────────── */
int conf_check_stale(const char *workdir, time_t stale_timeout_s)
{
if (stale_timeout_s == 0)
return 0;
char fname[256] = {0};
if (conf_read_current_filename(workdir, fname, sizeof(fname)) != 0)
return 0;
char ver_path[4096];
snprintf(ver_path, sizeof(ver_path), "%s/conf/%s", workdir, fname);
struct stat st;
if (stat(ver_path, &st) != 0)
return 0;
time_t age = time(NULL) - st.st_mtime;
return (age > stale_timeout_s) ? 1 : 0;
}
/* ── conf_trim_old_versions ─────────────────────── */
typedef struct {
char name[NAME_MAX];
time_t mtime;
} ver_entry_t;
static int ver_entry_cmp_desc(const void *a, const void *b)
{
const ver_entry_t *ea = (const ver_entry_t *)a;
const ver_entry_t *eb = (const ver_entry_t *)b;
if (eb->mtime > ea->mtime) return 1;
if (eb->mtime < ea->mtime) return -1;
return 0;
}
void conf_trim_old_versions(const char *workdir, int keep_n)
{
char conf_dir[4096];
make_conf_dir(workdir, conf_dir, sizeof(conf_dir));
/* Read current pointer to protect it from deletion */
char current_name[256] = {0};
if (conf_read_current_filename(workdir, current_name, sizeof(current_name)) != 0) {
/* No current pointer — nothing safe to trim without risking deleting the active file */
return;
}
ver_entry_t entries[256];
int n = 0;
DIR *d = opendir(conf_dir);
if (!d)
return;
struct dirent *e;
while ((e = readdir(d)) != NULL && n < 256) {
if (strncmp(e->d_name, "cluster.bin.", 12) != 0)
continue;
if (strstr(e->d_name, ".tmp") != NULL) {
char full[4096];
snprintf(full, sizeof(full), "%s/%s", conf_dir, e->d_name);
unlink(full);
continue;
}
char full[4096];
snprintf(full, sizeof(full), "%s/%s", conf_dir, e->d_name);
struct stat st;
if (stat(full, &st) == 0) {
strncpy(entries[n].name, e->d_name, NAME_MAX - 1);
entries[n].name[NAME_MAX - 1] = '\0';
entries[n].mtime = st.st_mtime;
n++;
}
}
closedir(d);
/* Sort descending by mtime (newest first) */
qsort(entries, (size_t)n, sizeof(entries[0]), ver_entry_cmp_desc);
/* Delete everything after keep_n, protecting the current pointer */
for (int i = keep_n; i < n; i++) {
if (strcmp(entries[i].name, current_name) == 0)
continue;
char full[4096];
snprintf(full, sizeof(full), "%s/%s", conf_dir, entries[i].name);
unlink(full);
}
}
/* ── config_contains_reserved_name ─────────────── */
int config_contains_reserved_name(cluster_config_t *cfg)
{
if (!cfg)
return 0;
/* Check queue names — xce_queue_t.queue is the queue name field */
for (int i = 0; i < cfg->nqueues; i++) {
if (cfg->queues[i] &&
cfg->queues[i]->queue &&
strcmp(cfg->queues[i]->queue, XCE_LOST_AND_FOUND_NAME) == 0)
return 1;
}
/* Check host config names — xce_hostconfig_t.name */
for (int i = 0; i < cfg->nhconfigs; i++) {
if (cfg->hconfigs[i] &&
cfg->hconfigs[i]->name &&
strcmp(cfg->hconfigs[i]->name, XCE_LOST_AND_FOUND_NAME) == 0)
return 1;
}
return 0;
}
/* ── load_primary_config_for_test ───────────────── */
#ifdef XCE_TESTING
int load_primary_config_for_test(cluster_t *c)
{
myconf_t *myconf = c->myconf;
char conf_dir[4096];
snprintf(conf_dir, sizeof(conf_dir), "%s/%s", myconf->workdir, myconf->name);
char fname[256] = {0};
int rc = conf_read_current_filename(conf_dir, fname, sizeof(fname));
if (rc != 0)
return rc;
free(myconf->conffile);
myconf->conffile = NULL;
if (asprintf(&myconf->conffile, "%s/conf/%s", conf_dir, fname) < 0)
return -ENOMEM;
return 0;
}
#endif
/* ── create_lost_and_found ──────────────────────── */
void create_lost_and_found(cluster_t *c)
{
if (!c->qmgr || !c->workermgr)
return;
#ifndef XCE_TESTING
/* key_find, create_queue, and makeWorker require librtti + queue.c which
* are not linked into the unit test binary. The !qmgr guard above means
* this block is unreachable in tests regardless. */
/* Synthetic queue — receives jobs whose queue was removed by a config
* upgrade. Created before persist_recover() so WAL replay can route
* unknown queue IDs here instead of failing. */
{
void *existing = NULL;
if (key_find(c->qmgr, (void *)XCE_LOST_AND_FOUND_NAME, &existing) <= 0) {
xce_queue_t lf = {0};
lf.queue = XCE_LOST_AND_FOUND_NAME;
queue_t *q = create_queue(c, &lf);
if (q)
INFO("Created synthetic 'lost_and_found' queue");
else
ERROR("Failed to create synthetic 'lost_and_found' queue");
}
}
/* Synthetic host — receives jobs whose host was removed by a config
* upgrade. Uses makeWorker directly; no hostconfig entry is required
* for this sentinel host. */
{
void *existing = NULL;
if (key_find(c->workermgr, (void *)XCE_LOST_AND_FOUND_NAME, &existing) <= 0) {
hostattrs_t lf_attrs = { .hosttype = XCE_LOST_AND_FOUND_NAME,
.hostmodel = XCE_LOST_AND_FOUND_NAME };
xce_worker_t lf_host = { .name = XCE_LOST_AND_FOUND_NAME,
.attrs = &lf_attrs };
worker_t *w = makeWorker(c->workermgr, &lf_host);
if (w)
INFO("Created synthetic 'lost_and_found' host");
else
ERROR("Failed to create synthetic 'lost_and_found' host");
}
}
#endif /* XCE_TESTING */
}
/* ── handle_config_request ──────────────────────── */
int handle_config_request(cluster_t *c, endpoint_t *ep, xce_request_t *req)
{
uint64_t their_tag = req->body.replconfigrequestmsg->version_tag;
uint64_t my_tag = c->myconf->config_version_tag;
xce_reply_t reply;
init_reply_by_req(&reply, req);
reply.data_which = RequestReply_data_replconfigreplymsg;
repl_config_reply_t *rmsg = (repl_config_reply_t *)calloc(
1, sizeof(repl_config_reply_t));
if (!rmsg) {
reply.status = -ENOMEM;
reply.data_which = RequestReply_data_nodata;
xce_reply(ep, c->myhost, &reply);
return -ENOMEM;
}
reply.data.replconfigreplymsg = rmsg;
if (their_tag == my_tag) {
reply.status = 0;
rmsg->up_to_date = 1;
rmsg->version_tag = 0;
rmsg->sha256 = NULL;
rmsg->bytes = NULL;
xce_reply(ep, c->myhost, &reply);
free(rmsg);
reply.data.replconfigreplymsg = NULL;
return 0;
}
/* Read raw config bytes */
uint8_t *bytes = NULL;
size_t blen = 0;
int rc = conf_read_raw(c->myconf->conffile, &bytes, &blen);
if (rc != 0) {
ERROR("handle_config_request: conf_read_raw failed: %d", rc);
reply.status = rc;
reply.data_which = RequestReply_data_nodata;
free(rmsg);
reply.data.replconfigreplymsg = NULL;
xce_reply(ep, c->myhost, &reply);
return rc;
}
/* Verify SHA-256 matches what we cached at startup */
uint8_t recomputed[32];
SHA256(bytes, blen, recomputed);
if (memcmp(recomputed, c->myconf->config_sha256, 32) != 0) {
ERROR("handle_config_request: config file changed since startup"
" — refusing to send (hash mismatch)");
free(bytes);
free(rmsg);
reply.data.replconfigreplymsg = NULL;
reply.status = -EIO;
reply.data_which = RequestReply_data_nodata;
xce_reply(ep, c->myhost, &reply);
return -EIO;
}
int rc2 = xce_replconfig_reply_raw(ep, c->myhost, req->seq,
my_tag,
recomputed, 32,
bytes, blen);
free(bytes);
/* rmsg is no longer needed — xce_replconfig_reply_raw builds capnp directly */
free(rmsg);
reply.data.replconfigreplymsg = NULL;
if (rc2 != 0)
ERROR("handle_config_request: xce_replconfig_reply_raw failed: %d", rc2);
return rc2;
}
/* ── handle_repl_config_push ────────────────────────────────────────────────
*
* Standby handler for OPCode_replConfigPush sent by the primary.
*
* Steps:
* 1. Extract raw bytes + sha256 from the pre-decoded request
* 2. Verify SHA-256 integrity
* 3. Save to disk via conf_save_config
* 4. Reload cluster_config in memory via load_primary_config
* 5. Send ReplConfigPushReply with accepted=1
*/
int handle_repl_config_push(cluster_t *c, endpoint_t *ep, xce_request_t *req)
{
if (req->body_which != Request_body_replconfigpushrequestmsg
|| !req->body.replconfigpushrequestmsg) {
WARN("handle_repl_config_push: malformed request");
return -EINVAL;
}
/* Req 1.3 (ha-heartbeat-liveness): any msg from primary resets standby
* liveness and steps down from CTRL_ELECTING (live primary discovered). */
{
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(
&c->ha.role, memory_order_acquire);
if (role == CTRL_STANDBY)
ha_on_heartbeat_received(&c->ha);
else if (role == CTRL_ELECTING)
ha_discover_primary(&c->ha, c);
}
repl_config_push_req_t *rcpr = req->body.replconfigpushrequestmsg;
if (!rcpr->raw_bytes || rcpr->raw_bytes_len == 0 || !rcpr->raw_sha256) {
WARN("handle_repl_config_push: missing bytes or sha256 (tag=%016" PRIx64 ")",
rcpr->version_tag);
return -EINVAL;
}
/* Verify transit SHA-256 */
uint8_t computed[32];
SHA256(rcpr->raw_bytes, rcpr->raw_bytes_len, computed);
if (memcmp(computed, rcpr->raw_sha256, 32) != 0) {
WARN("handle_repl_config_push: SHA-256 mismatch — rejecting push");
return -EBADMSG;
}
/* ha-term-safety task 4.3 (remediation r1, Finding A): SAVE the pushed bytes
* to disk out-of-band, but do NOT activate them here. config_activate_committed
* is the SOLE activation point — it loads the bytes into the live config ONLY
* after hash-verifying them against the latest committed WAL_CONFIG_MARKER
* (R7.5/R7.8). A pre-emptive load_primary_config here would make the just-pushed
* bytes LIVE before the gate runs, turning verify-before-activate into an
* after-the-fact overlay. The bytes travel out-of-band via this transport and
* NEVER inside a WAL payload (R7.4/R7.5). */
int rc = conf_save_config(c->myconf, rcpr->raw_bytes, rcpr->raw_bytes_len,
rcpr->raw_sha256);
if (rc != 0) {
WARN("handle_repl_config_push: conf_save_config failed: %d", rc);
return rc;
}
INFO("handle_repl_config_push: config bytes saved out-of-band "
"(pushed tag=%016" PRIx64 ") — activation gated on committed marker",
rcpr->version_tag);
/* Key activation to the committed marker. On mismatch (the marker references
* different/later bytes than the ones just pushed) config_activate_committed
* sets the loud non-serving serve-fence and returns -EAGAIN; the member STAYS
* ALIVE and a later config-push / commit-advance retries (R7.9). -ENOENT
* (no committed marker yet — first boot/bootstrap) falls back to loading the
* locally-held bytes; a successful verify activates. */
int act_rc = config_activate_committed(c);
if (act_rc != 0 && act_rc != -EAGAIN)
WARN("handle_repl_config_push: config_activate_committed rc=%d", act_rc);
/* ACK the primary */
repl_config_push_reply_t reply_body = {
.accepted = 1,
.version_tag = c->myconf->config_version_tag,
};
xce_reply_t reply;
init_reply_by_req(&reply, req);
reply.status = 0;
reply.data_which = RequestReply_data_replconfigpushreplymsg;
reply.data.replconfigpushreplymsg = &reply_body;
xce_reply(ep, c->myhost, &reply);
return 0;
}
/* ── send_upgrade_error ─────────────────────────── */
/* status: the negative errno carried to the requester (task 5.2 r1,
* reviewer finding 3: validation failures keep -EINVAL; the barrier
* failure path forwards its own rc, e.g. -ETIMEDOUT). */
static void send_upgrade_error(endpoint_t *ep, xce_request_t *req,
cluster_t *c, int status, const char *msg)
{
xce_reply_t reply;
init_reply_by_req(&reply, req);
reply.status = (status < 0) ? status : -EINVAL;
reply.data_which = RequestReply_data_nodata;
xce_reply(ep, c->myhost, &reply);
ERROR("OP_upgradeConfig rejected (status=%d): %s", reply.status, msg);
}
/* ── push_config_to_standbys ─────────────────────── */
/* push_ctx_t is exposed in config_repl.h under XCE_TESTING so tests can
* construct synthetic contexts. In production builds, define it locally. */
#ifndef XCE_TESTING
typedef struct {
endpoint_t *ep;
xce_peerinfo_t *from;
uint64_t new_tag;
const uint8_t *bytes;
size_t blen;
const uint8_t *sha256;
int acked;
int started;
int done; /* set by push_thread on exit */
} push_ctx_t;
#endif
/* push_thread_send_hook: test-only seam (same pattern as
* xce_wal_stream_request_hook) letting a unit test intercept/park the
* per-standby send so it can probe endpoint_ref/unref bracketing around a
* concurrent teardown (bug-1381) without a real transport connection. NULL
* in production — push_thread falls through to the real
* xce_replconfigpush_req_raw. */
int (*push_thread_send_hook)(endpoint_t *ep, xce_peerinfo_t *from,
uint64_t new_tag, const uint8_t *sha256,
const uint8_t *bytes, size_t blen) = NULL;
static void *push_thread(void *arg)
{
push_ctx_t *ctx = (push_ctx_t *)arg;
int rc = (push_thread_send_hook != NULL)
? push_thread_send_hook(ctx->ep, ctx->from, ctx->new_tag,
ctx->sha256, ctx->bytes, ctx->blen)
: xce_replconfigpush_req_raw(
ctx->ep, ctx->from,
ctx->new_tag,
ctx->sha256, 32,
ctx->bytes, ctx->blen);
if (rc == 0)
ctx->acked = 1;
else
WARN("push_config_to_standbys: push to endpoint %p failed: %d", ctx->ep, rc);
/* bug-1381: drop the ref push_config_transport_only took on ctx->ep before
* spawning us (the general bug-1298 refcount system) — this thread is the
* one non-owner reader of ep for the full push, so it owns the matching
* release. Last use of ctx->ep in this thread; safe to unref now. */
endpoint_unref(ctx->ep);
ctx->done = 1;
return NULL;
}
/* wait_all_with_timeout: join all push threads within a CLOCK_MONOTONIC deadline.
*
* Polls until all started threads have set ctx->done or the deadline expires,
* then signals *done = 1 (for future callers that check an early-exit flag),
* and joins all started threads (they exit quickly once ctx->done is set).
*
* Returns the number of threads that had acked before the deadline.
*/
static int wait_all_with_timeout(pthread_t *threads, push_ctx_t *ctxs,
int n, uint32_t timeout_ms, atomic_int *done)
{
struct timespec deadline;
clock_gettime(CLOCK_MONOTONIC, &deadline);
deadline.tv_sec += (long)(timeout_ms / 1000);
deadline.tv_nsec += (long)(timeout_ms % 1000) * 1000000L;
if (deadline.tv_nsec >= 1000000000L) {
deadline.tv_sec++;
deadline.tv_nsec -= 1000000000L;
}
/* Poll until deadline or all started threads have set their done flag */
for (;;) {
struct timespec now;
clock_gettime(CLOCK_MONOTONIC, &now);
if (now.tv_sec > deadline.tv_sec ||
(now.tv_sec == deadline.tv_sec &&
now.tv_nsec >= deadline.tv_nsec))
break;
int pending = 0;
for (int i = 0; i < n; i++) {
if (ctxs[i].started && !ctxs[i].done)
pending++;
}
if (pending == 0)
break;
usleep(5000); /* 5ms poll */
}
atomic_store(done, 1); /* signal threads that deadline has passed */
int acked = 0;
for (int i = 0; i < n; i++) {
if (ctxs[i].started) {
pthread_join(threads[i], NULL);
if (ctxs[i].acked)
acked++;
}
}
return acked;
}
#ifdef XCE_TESTING
int wait_all_with_timeout_for_test(pthread_t *threads, push_ctx_t *ctxs,
int n, uint32_t timeout_ms, atomic_int *done)
{
return wait_all_with_timeout(threads, ctxs, n, timeout_ms, done);
}
#endif
/* push_config_transport_only: transport fan-out half of
* push_config_to_standbys (exported — task 5.2 remediation r1): spawn one
* push thread per registered standby endpoint and wait for the sends.
* Best-effort streaming, NO barrier wait — the sends are non-blocking
* endpoint_enq()s, so this is safe on iomgr-reachable paths (election
* promotion, heartbeat-driven convergence, controller join). Alloc or
* thread failures degrade to fewer pushes, never to a fabricated quorum
* (the barrier wait in push_config_to_standbys owns the quorum verdict).
* bug-1381 (CLOSED): each standby ep is endpoint_ref()'d under c->ha.lock
* before the pointer is snapshotted into ctxs[], matched by exactly one
* endpoint_unref() (from push_thread, or from this function's own cleanup
* if the thread never started) — so a concurrent RDHUP teardown can drop
* the OWNER's reference but never frees ep out from under an in-flight
* push_thread.
* Stores the registered standby count in *n_out (may be NULL); returns
* the number of standbys that ACKed the transport push. */
int push_config_transport_only(cluster_t *c, const uint8_t *bytes,
size_t blen, const uint8_t sha256[32],
uint64_t new_tag, int *n_out)
{
int n;
int acked;
if (n_out)
*n_out = 0;
if (!c->ha.standby_endpoints)
return 0;
pthread_mutex_lock(&c->ha.lock);
n = c->ha.standby_endpoints->count;
if (n_out)
*n_out = n;
if (n == 0) {
pthread_mutex_unlock(&c->ha.lock);
return 0;
}
push_ctx_t *ctxs = calloc((size_t)n, sizeof(push_ctx_t));
if (!ctxs) {
pthread_mutex_unlock(&c->ha.lock);
WARN("push_config_transport_only: ctx alloc failed (n=%d)"
" — skipping transport push", n);
return 0;
}
for (int i = 0; i < n; i++) {
ctxs[i].ep = (endpoint_t *)zarray_get(c->ha.standby_endpoints, i);
ctxs[i].from = c->myhost;
ctxs[i].new_tag = new_tag;
ctxs[i].bytes = bytes;
ctxs[i].blen = blen;
ctxs[i].sha256 = sha256;
/* bug-1381: take our own reference on each standby ep while still
* holding c->ha.lock — the same window the pointer is snapshotted —
* mirroring the snapshot-lane fix (bug-1197/bug-1376). Without this,
* a concurrent RDHUP teardown (handle_endpoint_error -> zarray_del ->
* endpoint_killer -> endpoint_unref) can free ep while push_thread
* still reads/writes it for up to CONF_PUSH_TIMEOUT_MS. Every ref
* taken here has exactly one matching endpoint_unref: from push_thread
* on success, or from the cleanup below if the thread never started. */
endpoint_ref(ctxs[i].ep);
}
pthread_mutex_unlock(&c->ha.lock);
pthread_t *threads = malloc((size_t)n * sizeof(pthread_t));
if (!threads) {
for (int i = 0; i < n; i++)
endpoint_unref(ctxs[i].ep);
free(ctxs);
WARN("push_config_transport_only: thread alloc failed (n=%d)"
" — skipping transport push", n);
return 0;
}
for (int i = 0; i < n; i++) {
ctxs[i].started = (pthread_create(&threads[i], NULL, push_thread, &ctxs[i]) == 0);
if (!ctxs[i].started) {
WARN("push_config_transport_only: pthread_create failed for standby %d", i);
/* No thread will run push_thread's matching endpoint_unref —
* release our ref right here (bug-1381). */
endpoint_unref(ctxs[i].ep);
}
}
atomic_int push_done;
atomic_init(&push_done, 0);
acked = wait_all_with_timeout(threads, ctxs, n,
CONF_PUSH_TIMEOUT_MS, &push_done);
free(ctxs);
free(threads);
return acked;
}
int push_config_to_standbys(cluster_t *c, const uint8_t *bytes, size_t blen,
const uint8_t sha256[32], uint64_t new_tag,
int *acked_out)
{
/* Returns the majority-ACK barrier result (0 / -ETIMEDOUT / -EINVAL);
* the per-standby transport push count travels via acked_out (task 5.2
* re-plumb). The transport fan-out is best-effort — push failures and
* an empty standby set fall through to the barrier wait, which owns
* the quorum policy (n_peers <= 1 → immediate success; multi-member
* with no covering acks → -ETIMEDOUT; no n_active shortcut).
* VTASK-CONTEXT ONLY (task 5.2 remediation r1): the barrier blocks up
* to CONF_PUSH_TIMEOUT_MS — iomgr-reachable callers must use
* push_config_transport_only instead (see config_repl.h). */
int n = 0;
int acked = push_config_transport_only(c, bytes, blen, sha256, new_tag, &n);
if (acked_out)
*acked_out = acked;
/* Req 8.1–8.3 / barrier-rollout Req 4.1-4.3: wait for majority-quorum
* OP_upgradeConfigAck, passing the in-flight push tag for per-push
* evaluation (entry_lsn carries the tag — task 5.2). On quorum:
* advance committed_config_tag atomically (Req 8.5 / 4.3 — the
* advance stays inside this success branch only).
*
* ha-term-safety task 4.2: committed_config_tag is now a NON-AUTHORITATIVE
* CACHE. The authoritative committed configuration is DERIVED from the
* latest committed WAL_CONFIG_MARKER (config_committed_marker, R7.7); this
* store only keeps legacy consumers (status, heartbeat) consistent with the
* derived value. The marker is the ordering/consensus record; this barrier
* remains the out-of-band byte-delivery quorum. */
int rc = repl_wait_majority_ack(&c->persist.repl, new_tag,
CONF_PUSH_TIMEOUT_MS,
(uint8_t)OPCode_upgradeConfigAck);
if (rc == 0) {
atomic_store_explicit(&c->ha.committed_config_tag, new_tag,
memory_order_release);
INFO("Config committed on majority (tag=%016" PRIx64
", pushed=%d/%d standbys)", new_tag, acked, n);
} else {
WARN("Config push: no majority ACK (rc=%d, tag=%016" PRIx64
", pushed=%d/%d standbys)", rc, new_tag, acked, n);
}
return rc;
}
/* ── handle_upgrade_config ──────────────────────── */
/* Test-only seam (config-upgrade-snapshot task 7): fires in
* do_upgrade_config_with_bytes right after config_commit_change() succeeds
* -- the configuration-change marker has just reached the committed
* prefix, but materialization on every member (apply_committed_entry's
* WAL_CONFIG_MARKER case, persist.c) has not necessarily completed
* anywhere yet. This is exactly the window a later regression test needs a
* deterministic way to land a forced leadership change in (Requirements
* 4.3, 5.4, 6.3). NULL in production -- no behaviour change. Mirrors
* xce_snapshot_chunk_hook's shape. Declared extern in the test file (same
* pattern as xce_snapshot_chunk_hook). */
#ifdef XCE_TESTING
int (*xce_upgrade_marker_committed_hook)(cluster_t *c,
uint64_t version_tag) = NULL;
#endif
/*
* do_upgrade_config_with_bytes: core upgrade logic called after the raw
* capnp bytes have been extracted from the wire via
* xce_upgrade_config_req_extract_raw().
*
* Steps:
* 0. Validate size limits
* 1. Verify transit SHA-256
* 2. Decode into cluster_config_t via temp file + load_cluster_config_from_file
* 3. Reject if reserved name present
* 4. Increment generation, re-encode with updated counter
* 5. Save to disk atomically
* 6. Update in-memory generation
* 7. Push to standbys and wait for the majority-ACK barrier; on failure
* reply with send_upgrade_error and return (no success ACK, no restart)
* 8. ACK admin tool (success ⇔ majority-acked config)
*
* The function's job ends at step 8. Materialization of the new
* configuration on every member -- including this one -- happens
* asynchronously and independently via each member's own apply of the
* committed WAL_CONFIG_MARKER (apply_committed_entry, persist.c,
* config-upgrade-snapshot task 5). No restart is requested or required.
*/
static int do_upgrade_config_with_bytes(cluster_t *c, endpoint_t *ep,
xce_request_t *req,
const uint8_t *in_bytes,
size_t in_len,
const uint8_t in_sha[32])
{
/* 0. Validate size */
if (in_len == 0 || in_bytes == NULL || in_sha == NULL) {
send_upgrade_error(ep, req, c, -EINVAL, "Missing bytes or sha256 field");
return -EINVAL;
}
if (in_len > CONF_MAX_SIZE) {
send_upgrade_error(ep, req, c, -EINVAL, "Config blob exceeds CONF_MAX_SIZE");
return -EINVAL;
}
/* 1. Verify transit SHA-256 */
uint8_t computed[32];
SHA256(in_bytes, in_len, computed);
if (memcmp(computed, in_sha, 32) != 0) {
send_upgrade_error(ep, req, c, -EINVAL,
"SHA-256 mismatch -- blob corrupted in transit");
return -EINVAL;
}
/* 2. Write to a temp file and decode via load_cluster_config_from_file().
* This reuses the same decode path as the normal config load and avoids
* direct in-memory capnp buffer manipulation. */
char tmp_path[4096];
snprintf(tmp_path, sizeof(tmp_path), "/tmp/xce_upgrade_cfg.%lu.tmp",
(unsigned long)pthread_self());
int fd = open(tmp_path, O_WRONLY | O_CREAT | O_TRUNC, 0600);
if (fd < 0) {
send_upgrade_error(ep, req, c, -EINVAL, "Failed to create temp decode file");
return -errno;
}
if (write(fd, in_bytes, in_len) != (ssize_t)in_len) {
close(fd);
unlink(tmp_path);
send_upgrade_error(ep, req, c, -EINVAL, "Failed to write temp decode file");
return -EIO;
}
close(fd);
cluster_config_t *new_cfg = NULL;
int rc = load_cluster_config_from_file(tmp_path, &new_cfg);
unlink(tmp_path);
if (rc != 0 || !new_cfg) {
if (new_cfg) {
free_ClusterConfig(new_cfg);
free(new_cfg);
}
send_upgrade_error(ep, req, c, -EINVAL, "Failed to parse Cap'n Proto blob");
return -EINVAL;
}
/* 3. Reject reserved names */
if (config_contains_reserved_name(new_cfg)) {
free_ClusterConfig(new_cfg);
free(new_cfg);
send_upgrade_error(ep, req, c, -EINVAL,
"Config contains reserved name 'lost_and_found'");
return -EINVAL;
}
/* 4. Increment generation, re-encode */
uint64_t new_gen = c->myconf->config_generation + 1;
new_cfg->config_generation = new_gen;
struct capn out_cap;
capn_init_malloc(&out_cap);
struct capn_segment *seg = capn_root(&out_cap).seg;
ClusterConfig_ptr out_p;
encode_ClusterConfig_ptr(seg, &out_p, new_cfg);
capn_setp(capn_root(&out_cap), 0, out_p.p);
free_ClusterConfig(new_cfg);
free(new_cfg);
new_cfg = NULL;
size_t out_len = capn_size(&out_cap);
uint8_t *out_bytes = (uint8_t *)malloc(out_len);
if (!out_bytes) {
capn_free(&out_cap);
send_upgrade_error(ep, req, c, -EINVAL, "OOM re-encoding config");
return -ENOMEM;
}
capn_write_mem(&out_cap, out_bytes, out_len, 0);
capn_free(&out_cap);
SHA256(out_bytes, out_len, computed);
/* 5. Save to disk */
rc = conf_save_config(c->myconf, out_bytes, out_len, computed);
if (rc != 0) {
free(out_bytes);
send_upgrade_error(ep, req, c, -EINVAL, "Failed to save new config to disk");
return rc;
}
/* 6. Update in-memory generation */
c->myconf->config_generation = new_gen;
/* 7. Commit the config change through the unified (term,lsn) log
* (ha-term-safety task 4.2): config_commit_change appends a
* WAL_CONFIG_MARKER as the CONSENSUS RECORD, pushes the bytes out-of-band,
* and waits for the MARKER to reach the committed prefix (the majority
* requirement is now "marker committed" — R7.1/R7.11). The committed config
* is DERIVED from that marker (config_committed_marker); committed_config_tag
* is only a cache refreshed inside the success branch. A non-zero rc means
* the marker never committed, so the requester gets an explicit error and
* the cluster stays on the old committed config (no success ACK, no
* restart). Disk save and generation bump already happened (pre-existing
* sequence); convergence of a half-pushed config remains the OP_replConfig
* anti-entropy behavior. */
int acked = 0;
rc = config_commit_change(c, out_bytes, out_len, computed,
c->myconf->config_version_tag, &acked);
free(out_bytes);
if (rc != 0) {
/* Forward the barrier rc (-ETIMEDOUT / -EINVAL) as the reply
* status so the admin tool sees WHY the commit failed (task 5.2
* r1, reviewer finding 3). */
send_upgrade_error(ep, req, c, rc,
"Config marker failed to reach committed prefix"
" -- upgrade not committed, no restart");
return rc;
}
#ifdef XCE_TESTING
if (xce_upgrade_marker_committed_hook)
(void)xce_upgrade_marker_committed_hook(c, c->myconf->config_version_tag);
#endif
/* 8. ACK the admin tool (success ⇔ majority-acked config) */
upgrade_config_reply_t urep;
memset(&urep, 0, sizeof(urep));
urep.version_tag = c->myconf->config_version_tag;
urep.acked_count = (uint32_t)acked;
urep.error = NULL;
xce_reply_t reply;
init_reply_by_req(&reply, req);
reply.status = 0;
reply.data_which = RequestReply_data_upgradeconfigreplymsg;
reply.data.upgradeconfigreplymsg = &urep;
xce_reply(ep, c->myhost, &reply);
/* Materialization now happens via apply_committed_entry's
* WAL_CONFIG_MARKER case (persist.c, config-upgrade-snapshot task 5) on
* every member's own apply of this marker -- no restart needed. */
INFO("Config upgraded to tag=%016" PRIx64 " gen=%" PRIu64
" (acked=%d standbys)",
c->myconf->config_version_tag, new_gen, acked);
return 0;
}
#if CONF_ENABLE_UPGRADE_CONFIG
int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req)
{
/* xce_decode() extracted sha256 + bytes from the wire into
* req->body.upgradeconfigrequestmsg->raw_sha256 / raw_bytes before
* capn_free() ran. Use them directly — no raw packet copy needed. */
uint8_t *in_bytes = NULL;
size_t in_len = 0;
uint8_t *in_sha = NULL;
int rc = xce_upgrade_config_req_extract(req, &in_bytes, &in_len, &in_sha);
if (rc != 0 || !in_bytes || !in_sha) {
send_upgrade_error(ep, req, c, -EINVAL,
"Failed to extract bytes from UpgradeConfigRequest");
return rc != 0 ? rc : -EINVAL;
}
if (in_len == 0) {
send_upgrade_error(ep, req, c, -EINVAL, "bytes or sha256 field missing");
return -EINVAL;
}
/* in_bytes and in_sha are owned by req — do NOT free here */
return do_upgrade_config_with_bytes(c, ep, req, in_bytes, in_len, in_sha);
}
#else
int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req)
{
(void)c;
xce_reply_t reply;
init_reply_by_req(&reply, req);
reply.status = -ENOSYS;
reply.data_which = RequestReply_data_nodata;
xce_reply(ep, c->myhost, &reply);
return -ENOSYS;
}
#endif
/* ── xce_probe_config_tag ───────────────────────────────────────────────────
*
* Single-RTT config tag probe to one peer controller over SSL.
*
* Wire format:
* Send: uint64_t size_hdr + capnp(ProbeConfigTagRequest) [direct struct]
* Receive: uint64_t size_hdr + capnp(ProbeConfigTagReply) [direct struct]
*
* Returns:
* 0 — success; *tag_out holds peer's config_tag (0 = no config)
* -ECONNREFUSED — peer refused the TCP connection (port not open)
* -ETIMEDOUT — connect poll timed out
* other negative errno — transient failure
*
* Requirements: 3.1, 1.3
* Boundary: Config Bootstrap (config_repl.c)
*/
int xce_probe_config_tag(const char *host, uint16_t port, uint64_t *tag_out)
{
if (!host || !tag_out)
return -EINVAL;
*tag_out = 0;
SSL *ssl = NULL;
int sock = -1;
int flags_saved = 0;
/* ── Step 1: TCP connect (+ optional SSL handshake) ── */
{
struct addrinfo hints, *res = NULL;
memset(&hints, 0, sizeof(hints));
hints.ai_family = AF_UNSPEC;
hints.ai_socktype = SOCK_STREAM;
char port_str[16];
snprintf(port_str, sizeof(port_str), "%u", (unsigned)port);
if (getaddrinfo(host, port_str, &hints, &res) != 0)
return -EHOSTUNREACH;
sock = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
if (sock < 0) { int e = errno; freeaddrinfo(res); return -e; }
flags_saved = fcntl(sock, F_GETFL, 0);
fcntl(sock, F_SETFL, flags_saved | O_NONBLOCK);
connect(sock, res->ai_addr, res->ai_addrlen);
freeaddrinfo(res);
struct pollfd pfd = { .fd = sock, .events = POLLOUT | POLLERR | POLLHUP };
int pr;
do { pr = poll(&pfd, 1, 500); } while (pr < 0 && errno == EINTR);
if (pr == 0) { close(sock); return -ETIMEDOUT; }
if (pr < 0 || (pfd.revents & (POLLERR | POLLHUP))) { close(sock); return -ECONNREFUSED; }
int err = 0; socklen_t el = sizeof(err);
getsockopt(sock, SOL_SOCKET, SO_ERROR, &err, &el);
if (err) { close(sock); return -err; }
fcntl(sock, F_SETFL, flags_saved & ~O_NONBLOCK);
/* Optional SSL handshake — only when s_probe_ssl_ctx is set */
SSL_CTX *ssl_ctx = (SSL_CTX *)s_probe_ssl_ctx;
if (ssl_ctx) {
ssl = SSL_new(ssl_ctx);
if (!ssl) { close(sock); return -ENOMEM; }
SSL_set_fd(ssl, sock);
int sr;
do {
sr = SSL_connect(ssl);
if (sr <= 0) {
int ssl_err = SSL_get_error(ssl, sr);
if (ssl_err != SSL_ERROR_WANT_READ &&
ssl_err != SSL_ERROR_WANT_WRITE)
break;
}
} while (sr <= 0);
if (sr <= 0) {
SSL_free(ssl);
ssl = NULL;
close(sock);
return -ECONNREFUSED;
}
}
}
/* ── Step 2: Encode + send ProbeConfigTagRequest (direct struct) ── */
{
struct capn c;
struct capn_segment *cs;
struct ProbeConfigTagRequest req_s;
ProbeConfigTagRequest_ptr req_p;
uint8_t *buf = NULL;
int64_t blen = 0;
capn_init_malloc(&c);
cs = capn_root(&c).seg;
memset(&req_s, 0, sizeof(req_s));
req_s.name.str = ""; /* anonymous probe; peer logs remote addr */
req_s.name.len = 0;
req_p = new_ProbeConfigTagRequest(cs);
write_ProbeConfigTagRequest(&req_s, req_p);
capn_setp(capn_root(&c), 0, req_p.p);
blen = capn_size(&c);
if (blen > 0)
buf = (uint8_t *)malloc((size_t)blen);
if (!buf || capn_write_mem(&c, buf, (size_t)blen, 0) < 0) {
free(buf);
capn_free(&c);
close(sock);
return -EIO;
}
capn_free(&c);
/* Wire framing: 8-byte size prefix then capnp payload (matches ha.c pattern) */
uint64_t hdr = (uint64_t)blen;
ssize_t nsent;
if (ssl) {
nsent = (ssize_t)SSL_write(ssl, &hdr, (int)sizeof(hdr));
if (nsent == (ssize_t)sizeof(hdr))
nsent = (ssize_t)SSL_write(ssl, buf, (int)blen);
} else {
nsent = (ssize_t)write(sock, &hdr, sizeof(hdr));
if (nsent == (ssize_t)sizeof(hdr))
nsent = (ssize_t)write(sock, buf, (size_t)blen);
}
free(buf);
if (nsent != (ssize_t)blen) {
if (ssl) { SSL_shutdown(ssl); SSL_free(ssl); }
close(sock);
return -EIO;
}
}
/* ── Step 3: Read + decode ProbeConfigTagReply (direct struct) ── */
{
uint64_t hdr = 0;
size_t off = 0;
/* Read 8-byte size header */
while (off < sizeof(hdr)) {
int n = ssl ? SSL_read(ssl, (char *)&hdr + off, (int)(sizeof(hdr) - off))
: (int)read(sock, (char *)&hdr + off, sizeof(hdr) - off);
if (n <= 0) {
if (ssl) { SSL_shutdown(ssl); SSL_free(ssl); }
close(sock);
return -EIO;
}
off += (size_t)n;
}
if (hdr == 0 || hdr > 65536) { /* sanity cap — reply is tiny */
if (ssl) { SSL_shutdown(ssl); SSL_free(ssl); }
close(sock);
return -EMSGSIZE;
}
uint8_t *buf = (uint8_t *)malloc((size_t)hdr);
if (!buf) {
if (ssl) { SSL_shutdown(ssl); SSL_free(ssl); }
close(sock);
return -ENOMEM;
}
off = 0;
while (off < (size_t)hdr) {
int n = ssl ? SSL_read(ssl, buf + off, (int)((size_t)hdr - off))
: (int)read(sock, buf + off, (size_t)hdr - off);
if (n <= 0) {
free(buf);
if (ssl) { SSL_shutdown(ssl); SSL_free(ssl); }
close(sock);
return -EIO;
}
off += (size_t)n;
}
struct capn c;
struct ProbeConfigTagReply rep_s;
ProbeConfigTagReply_ptr rep_p;
if (capn_init_mem(&c, buf, (size_t)hdr, 0) != 0) {
free(buf);
close(sock);
return -EPROTO;
}
rep_p.p = capn_getp(capn_root(&c), 0, 1);
read_ProbeConfigTagReply(&rep_s, rep_p);
capn_free(&c);
free(buf);
*tag_out = rep_s.configTag;
}
if (ssl) { SSL_shutdown(ssl); SSL_free(ssl); }
close(sock);
INFO("HA: probe config tag from %s:%u tag=%016" PRIx64,
host, (unsigned)port, *tag_out);
return 0;
}
#ifdef XCE_TESTING
/* Injection point for unit tests — NULL means use real xce_probe_config_tag */
int (*xce_probe_config_tag_hook)(const char *host, uint16_t port,
uint64_t *tag_out) = NULL;
#endif
/* ── sanitise_cn ────────────────────────────────── */
char *sanitise_cn(const char *src, size_t maxlen)
{
static char buf[128];
size_t n = maxlen < sizeof(buf) - 1 ? maxlen : sizeof(buf) - 1;
size_t i;
for (i = 0; i < n && src[i] != '\0'; i++)
buf[i] = (src[i] >= 0x20 && src[i] < 0x7f) ? src[i] : '?';
buf[i] = '\0';
return buf;
}
/* ── handle_ctrlhello ─────────────────────────────────────────────────────
*
* Connection setup only — Req 1.1, 1.2 (ha-repl-hardening).
*
* All replication decisions (config push, WAL streaming, snapshot transfer)
* are handled by the heartbeat decision loop (ha_process_standby_state).
* This handler only bookmarks the incoming connection so the standby's
* liveness watchdog starts ticking.
*
* Steps:
* 1. Validate request body
* 2. Set primary_ctrl_fd = ep->fd (first ctrlhello from winner, idempotent)
* 3. Call ha_on_heartbeat_received (resets liveness watchdog)
* 4. Arm ha_arm_heartbeat_timeout_vtask
*/
int handle_ctrlhello(cluster_t *c, endpoint_t *ep, xce_request_t *req)
{
/* 1. Validate request body */
if (req->body_which != Request_body_ctrlhellomsg || !req->body.ctrlhellomsg) {
WARN("handle_ctrlhello: malformed request (body_which=%d)", req->body_which);
return -EINVAL;
}
/* 2–4. Connection bookkeeping: designate this connection as the primary
* control channel on the standby side. Guard with primary_ctrl_fd == -1
* for idempotency (Req 1.1, 1.2). */
if (c && ep &&
(ctrl_role_t)atomic_load_explicit(&c->ha.role,
memory_order_acquire) == CTRL_STANDBY &&
c->ha.primary_ctrl_fd == -1) {
c->ha.primary_ctrl_fd = ep->fd;
ha_on_heartbeat_received(&c->ha);
ha_arm_heartbeat_timeout_vtask(&c->ha, c);
INFO("HA: primary control channel established (fd=%d)", ep->fd);
}
return 0;
}
/* ── Task 3.4: Batch WAL Sender ──────────────────────────────────────────────
*
* ha_stream_wal_to_standby: stream WAL entries [standby_lsn+1, committed_lsn]
* to a standby endpoint as batched OPCode_walCatchupEntry messages.
*
* Design (§Batch WAL Sender):
* - Read WAL segment files from wal_path in LSN order (seg-*.wal format).
* - Accumulate entries into a batch up to WAL_STREAM_BATCH_ENTRIES (256) or
* WAL_STREAM_BATCH_BYTES (65536) total payload bytes.
* - On batch full: enqueue via xce_request() (non-blocking), start new batch.
* - After all segments: flush any partial batch immediately (Req 5.4).
* - All xce_request() calls are enqueue-only — never blocks the caller thread.
*
* Under XCE_TESTING, xce_wal_stream_request_hook intercepts the per-batch
* xce_request() call so unit tests can count batches without a real endpoint.
*
* Requirements: 5.1, 5.2, 5.4, 5.5
*/
#ifdef XCE_TESTING
/* Test-only seam: intercepts each batch xce_request() call.
* Defined here (translation unit scope); declared extern in test file. */
int (*xce_wal_stream_request_hook)(endpoint_t *ep, xce_request_t *req) = NULL;
/* raft-ha-core-safety task 2.3 — test-only fault-injection seam for the catch-up
* streamer's entry-body read (the "Catch-up Payload Integrity" boundary). When
* non-NULL AND it returns nonzero for an entry, ha_stream_wal_to_standby treats
* that entry's body pread as a SHORT READ — the same outcome as a malloc-fail —
* which reproduces the live cross-term catch-up payload-loss signature: a
* non-empty source entry is shipped with PLEN=0. NULL in production: behaviour
* is then byte-for-byte unchanged. This seam ONLY makes the defect
* deterministically observable for the task-2.3 RED test; the fix that stops the
* empty ship (fail loud + bounded retry) is task 2.4. Declared extern in the
* test file (same pattern as xce_wal_stream_request_hook). */
int (*ha_stream_body_read_fault_hook)(uint64_t lsn, uint64_t stored_term,
uint64_t leader_term,
uint32_t payload_len) = NULL;
#endif
/* send_wal_batch: enqueue a single OPCode_walCatchupEntry batch message.
* Entry payloads (raw_payload/raw_payload_len) travel as capnp Data via
* xce_wal_batch_req_raw (task 7.3) — the generated encode path cannot
* carry raw bytes (xce_data_t limitation, same as replConfigPush).
* Does not free any memory — caller manages entry lifetime.
* Returns 0 on success. */
static int send_wal_batch(endpoint_t *ep,
wal_catchup_entry_t **entries, int n_entries)
{
if (n_entries <= 0)
return 0;
#ifdef XCE_TESTING
if (xce_wal_stream_request_hook != NULL) {
/* Build the batch container the hook expects */
wal_catchup_entry_t batch;
memset(&batch, 0, sizeof(batch));
batch.n_entries = n_entries;
batch.entries = entries;
xce_request_t req;
memset(&req, 0, sizeof(req));
req.opcode = OPCode_walCatchupEntry;
req.body_which = Request_body_walcatchupentrymsg;
req.body.walcatchupentrymsg = &batch;
return xce_wal_stream_request_hook(ep, &req);
}
#endif
if (ep)
xce_wal_batch_req_raw(ep, entries, n_entries);
return 0;
}
/* Maximum bounded re-attempts when an entry body read makes no forward progress
* (a zero/EOF/EINTR short read or an allocation failure). Small and finite so the
* loop can never busy-spin; a regular-file body completes in one or two reads, so
* this only ever spends attempts on a genuinely failing read. No sleep/backoff —
* a transient cause (EINTR, partial read) recovers on the next immediate attempt;
* a persistent cause exhausts the budget and is surfaced loud (never shipped
* empty). raft-ha-core-safety task 2.4. */
#define WAL_STREAM_BODY_READ_MAX_ATTEMPTS 4
/* stream_read_entry_body: read a WAL entry body of `payload_len` bytes at
* `body_off` into a freshly malloc'd buffer, accumulating partial reads and
* bounded-retrying a no-progress read. On success returns 0 with *out_buf set to
* a heap buffer the caller owns (free()); on persistent failure returns negative
* (-ENOMEM / -EIO) with *out_buf == NULL.
*
* raft-ha-core-safety task 2.4 — the SEND-side payload-integrity fix. The
* baseline freed the buffer and shipped the entry with PLEN=0 on a short
* pread/malloc-fail, silently transmitting a non-empty COMMITTED entry as empty
* (the live LSN-9 divergence). This helper instead fails LOUD so the caller can
* abort the stream rather than degrade the entry to empty. A short read is not
* inherently fatal: pread may legitimately return fewer bytes, so we re-read from
* the advanced offset (forward progress never spends an attempt); only a read
* that returns <= 0 consumes the bounded retry budget. */
static int stream_read_entry_body(int fd, uint32_t body_off,
uint32_t payload_len, uint64_t lsn,
uint64_t stored_term, uint64_t leader_term,
uint8_t **out_buf)
{
*out_buf = NULL;
uint8_t *pbuf = NULL;
for (int a = 0; a < WAL_STREAM_BODY_READ_MAX_ATTEMPTS && !pbuf; a++)
pbuf = malloc(payload_len);
if (!pbuf)
return -ENOMEM;
uint32_t got = 0;
int attempts = 0;
while (got < payload_len) {
ssize_t pr = pread(fd, pbuf + got, payload_len - got,
(off_t)(body_off + got));
#ifdef XCE_TESTING
/* raft-ha-core-safety task 2.3/2.4 fault-injection seam: a test may force
* this read to look SHORT (pr=0) to exercise the fail-loud + bounded-retry
* path. NULL in production — no behaviour change. The hook fires on every
* attempt, so a permanent fault exhausts the budget (fail loud) while a
* transient one (hook returns 0 after N) recovers. */
if (ha_stream_body_read_fault_hook != NULL &&
ha_stream_body_read_fault_hook(lsn, stored_term, leader_term,
payload_len))
pr = 0;
#endif
if (pr > 0) {
got += (uint32_t)pr; /* forward progress — does not spend an attempt */
continue;
}
/* pr <= 0: EOF/zero/EINTR/error — bounded retry, no busy-spin. */
if (++attempts >= WAL_STREAM_BODY_READ_MAX_ATTEMPTS) {
free(pbuf);
return -EIO;
}
}
*out_buf = pbuf;
return 0;
}
/* raft-ha-core-safety task 2.5 (send-side torn-read guard): read ONE WAL
* entry's header + (if payload_len>0) full body while holding p->lock, so this
* read observes a consistent snapshot with respect to a CONCURRENT write to
* the SAME file -- persist_wal_write_replicated (and the primary's own append
* path) write under p->lock, but this function's reads previously ran lock-
* free, so a node that is simultaneously receiving new entries from its own
* upstream leader AND streaming older entries downstream to a catching-up
* standby could observe the downstream read torn: a committed entry with a
* real non-empty payload on disk read back with a short/zero payload_len or a
* partially-written body, then faithfully shipped and persisted downstream as
* empty -- a genuine committed-prefix divergence at a historical LSN (not the
* already-guarded body-short-read case in stream_read_entry_body, which
* assumes a syntactically valid header; this closes the header-itself-torn
* gap). Held only for one entry's read, not the whole stream, so a long catch-
* up does not block writers throughout. p may be NULL only in count-only
* tests (no live writer to race against).
*
* Returns 0 on success (*hdr_out populated; *payload_out set iff
* hdr_out->payload_len > 0), -1 on a short/EOF header read (caller treats as
* "nothing more in this segment"), or stream_read_entry_body's negative rc on
* a body-read failure (caller aborts the stream, per its own fail-loud
* contract). */
static int read_one_wal_entry_locked(int fd, uint32_t off, const persist_t *p,
uint64_t leader_term, uint64_t standby_lsn,
wal_entry_hdr_t *hdr_out,
uint8_t **payload_out)
{
pthread_mutex_t *lk = p ? (pthread_mutex_t *)&p->lock : NULL;
*payload_out = NULL;
if (lk)
pthread_mutex_lock(lk);
ssize_t n = pread(fd, hdr_out, sizeof(*hdr_out), (off_t)off);
if (n < (ssize_t)sizeof(*hdr_out)) {
if (lk)
pthread_mutex_unlock(lk);
return -1;
}
int rc = 0;
/* Only entries actually being streamed (above standby_lsn) need their
* body read at all -- matches the caller's own send-vs-scan-only split,
* so this fix does not add wasted I/O for the already-acked prefix. */
if (hdr_out->payload_len > 0 && hdr_out->lsn > standby_lsn) {
/* p->lock is already held here -- persist_entry_term() would
* re-acquire it and self-deadlock; persist_entry_term_locked() is
* the lock-free counterpart for exactly this case. */
uint64_t stored_term = p ? persist_entry_term_locked(p, hdr_out->lsn)
: hdr_out->term;
rc = stream_read_entry_body(fd, off + (uint32_t)sizeof(*hdr_out),
hdr_out->payload_len, hdr_out->lsn,
stored_term, leader_term, payload_out);
}
if (lk)
pthread_mutex_unlock(lk);
return rc;
}
int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path,
const persist_t *p, uint64_t leader_term,
uint64_t standby_lsn, uint64_t committed_lsn)
{
if (!wal_path)
return -EINVAL;
if (standby_lsn >= committed_lsn)
return 0; /* nothing to stream */
/* Req 7.1: log WAL catch-up start with start/target LSN */
INFO("ha_stream_wal_to_standby: starting WAL catch-up "
"start_lsn=%" PRIu64 " target_lsn=%" PRIu64,
standby_lsn + 1, committed_lsn);
/* ── Enumerate WAL segment files, sort ascending by first LSN ── */
DIR *d = opendir(wal_path);
if (!d)
return 0; /* no WAL directory yet — treat as empty */
uint64_t seg_lsns[4096];
int nseg = 0;
struct dirent *ent;
while ((ent = readdir(d)) != NULL &&
nseg < (int)(sizeof(seg_lsns) / sizeof(*seg_lsns))) {
uint64_t sl = 0;
if (sscanf(ent->d_name, "seg-%" SCNx64 ".wal", &sl) != 1) continue;
seg_lsns[nseg++] = sl;
}
closedir(d);
/* Insertion sort ascending by segment start LSN */
for (int i = 1; i < nseg; i++) {
uint64_t key = seg_lsns[i];
int j = i - 1;
while (j >= 0 && seg_lsns[j] > key) { seg_lsns[j+1] = seg_lsns[j]; j--; }
seg_lsns[j+1] = key;
}
/* ── Accumulate and send batches ── */
wal_catchup_entry_t *batch_entries[WAL_STREAM_BATCH_ENTRIES];
int batch_n = 0;
size_t batch_bytes = 0;
/* ha-term-safety task 3.4: track the previous streamed entry's
* (lsn, term) so each catch-up entry carries the AppendEntries prev tuple
* (prev_log_index, prev_log_term) + its own term. The standby intake routes
* every term-carrying sub-entry through raft_on_append_entries_request so it
* persists the LEADER's term (not a local substitute). Entries are streamed
* in strict ascending LSN order, so the prev tuple of entry N is exactly the
* (lsn, term) of the entry emitted just before it (0/0 before the first). */
uint64_t prev_streamed_lsn = 0;
uint64_t prev_streamed_term = 0;
for (int si = 0; si < nseg; si++) {
/* Skip segments whose entire range is at or below standby_lsn */
if (si + 1 < nseg && seg_lsns[si + 1] <= standby_lsn)
continue;
/* Stop once the segment starts beyond committed_lsn */
if (seg_lsns[si] > committed_lsn)
break;
char seg_path[4096];
snprintf(seg_path, sizeof(seg_path),
"%s/seg-%016" PRIx64 ".wal", wal_path, seg_lsns[si]);
int fd = open(seg_path, O_RDONLY);
if (fd < 0) continue;
uint32_t off = 0;
for (;;) {
wal_entry_hdr_t hdr;
uint8_t *locked_pbuf = NULL;
int rrc0 = read_one_wal_entry_locked(fd, off, p, leader_term,
standby_lsn, &hdr,
&locked_pbuf);
if (rrc0 == -1) break; /* short/EOF header read */
if (hdr.lsn == 0 && hdr.type == 0 && hdr.payload_len == 0) {
free(locked_pbuf);
break;
}
if (hdr.lsn > committed_lsn) {
free(locked_pbuf);
break;
}
/* WAL_CHECKPOINT is a local-only marker (persist_write_checkpoint_
* marker's own doc comment: "replay_wal SKIPS checkpoint markers...
* the recovered index never holds a checkpoint LSN; the live index
* must match") -- it consumes a real LSN slot in THIS node's local
* current_lsn sequence but is never a logical Raft log entry and is
* never sent on the live broadcast path either
* (persist_wal_write_replicated rejects type==WAL_CHECKPOINT with
* -EINVAL). Skip it entirely -- do not build/send a catchup entry
* for it.
*
* bug-1475 (fix): DO fall through to the SAME unconditional
* prev-tracker advance every other scanned entry gets below
* (the "for EVERY scanned entry" logic a few lines down) instead
* of special-casing a skip here. A standby's own current_lsn/
* durable_floor ALSO advances past this exact LSN the moment it
* independently applies the SAME committed WAL_CONFIG_MARKER (any
* member that commits a marker immediately writes its own local
* checkpoint at the marker's own LSN+1, deterministically the
* SAME lsn on every member that has caught up through the
* marker) -- so the checkpoint's LSN is a real, held position on
* BOTH sides, not a gap. The OLD behavior (never advancing the
* tracker across it) under-counted the sender's prev_log_index by
* one for every checkpoint encountered, permanently diverging it
* from the receiver's own (now correctly co-tracked -- see
* ha_apply_term_entry_via_library's bug-1475 fix) log_len: the
* first post-checkpoint entry would spuriously "succeed" at the
* WRONG index, after which every following entry's prev_log_index
* permanently exceeded the receiver's true log length --
* ae_reject_logmismatch forever, with the primary re-streaming
* the identical doomed range every heartbeat. Falling through
* here (falling to the same code path as every other entry)
* fixes the divergence at its source.
*
* bug-1475 follow-up: this tracker advance alone was NOT
* sufficient — persist_entry_term(p, checkpoint_lsn) (the term
* this tracker reads for prev_log_term) turned out to be
* TIME-DEPENDENT before a further persist.c fix: it correctly
* read absent (0) right after the checkpoint, but silently
* flipped to WAL_LEGACY_TERM (1) the instant this leader
* appended any LATER real entry (which has already happened by
* the time this streamer runs), because persist.c's term-index
* array growth zero-filled the checkpoint's never-explicitly-
* recorded gap slot and read that zero-fill back as present. A
* receiver that had not yet appended past its own checkpoint
* still (correctly) read 0 for the same LSN, so logOk's
* prev_log_term equality check mismatched (0 != 1) and rejected
* forever. Fixed with a TERM_INDEX_UNSET sentinel in persist.c
* (record_entry_term/entry_term_locked) that keeps a
* never-recorded slot reading absent permanently, independent of
* later growth — see TERM_INDEX_UNSET's comment there. */
if (hdr.lsn > standby_lsn && hdr.type != WAL_CHECKPOINT) {
wal_catchup_entry_t *entry = calloc(1, sizeof(*entry));
if (!entry) {
free(locked_pbuf);
/* OOM: flush current batch and skip this entry */
if (batch_n > 0) {
send_wal_batch(ep, batch_entries, batch_n);
for (int bi = 0; bi < batch_n; bi++) {
free(batch_entries[bi]->raw_payload);
free(batch_entries[bi]);
}
batch_n = 0;
batch_bytes = 0;
}
/* Advance the prev tracker even on the OOM-skip so a later
* entry's prev tuple stays correct. bug-1028: term from the
* authoritative index, not raw hdr.term (see main tracker). */
prev_streamed_lsn = hdr.lsn;
prev_streamed_term = p ? persist_entry_term(p, hdr.lsn)
: hdr.term;
off += (uint32_t)(sizeof(hdr) + hdr.payload_len);
continue;
}
entry->lsn = hdr.lsn;
entry->entry_type = (uint8_t)hdr.type;
entry->n_entries = 0;
entry->entries = NULL;
entry->payload = NULL;
/* ha-term-safety task 3.4: carry the leader's term + prev tuple
* so the standby intake routes this through the library and
* persists the LEADER's term (R1.3/R1.4). prev_streamed_* holds
* the (lsn,term) of the immediately preceding WAL entry (0/0
* before the first), which IS the predecessor in contiguous LSN
* order. */
/* bug-1028: own term from the authoritative per-LSN index, not
* raw hdr.term. prev_streamed_term is already normalised above. */
entry->term = p ? persist_entry_term(p, hdr.lsn)
: hdr.term;
entry->prev_log_index = prev_streamed_lsn;
entry->prev_log_term = prev_streamed_term;
/* bug-1067: stamp the leader's CURRENT term as the AppendEntries
* header term. entry->term above is the entry's STORED term (may
* be a PRIOR leader's term for a historical catch-up entry); the
* standby intake must check the leader's current term, not the
* stored term, or it stale-rejects every cross-term entry. */
entry->leader_current_term = leader_term;
if (hdr.payload_len > 0) {
/* raft-ha-core-safety task 2.4/2.5: the body was already
* read (with bounded retry) by read_one_wal_entry_locked
* above, under the SAME lock as the header -- a non-empty
* COMMITTED entry must NEVER be shipped empty, whether from
* a short body read (task 2.4) or a header/body torn read
* racing a concurrent write to this same file (task 2.5). */
if (rrc0 != 0) {
/* FAIL LOUD: the body could not be read in full after the
* bounded retry. Do NOT degrade the entry to PLEN=0 —
* abort the stream here. Flush the valid prefix already
* batched (entries strictly below this LSN remain correct
* and contiguous), drop THIS entry, and return an error so
* the ACK-driven catch-up continuation re-streams from this
* LSN on the next cycle. The standby simply does not
* advance past the gap — it never receives a committed
* entry as empty. The controller stays alive and
* self-heals (CLAUDE.md process-lifecycle invariant:
* fail-loud = refuse to ship divergent state, not exit). */
ERROR("ha_stream_wal_to_standby: entry body read FAILED "
"for lsn=%" PRIu64 " payload_len=%u stored_term=%"
PRIu64 " leader_term=%" PRIu64 " rc=%d after %d "
"attempts — ABORTING stream, NOT shipping empty (a "
"non-empty committed entry must never be transmitted "
"as PLEN=0; catch-up will re-stream from this LSN)",
hdr.lsn, hdr.payload_len, entry->term, leader_term,
rrc0, WAL_STREAM_BODY_READ_MAX_ATTEMPTS);
free(entry);
free(locked_pbuf);
if (batch_n > 0) {
send_wal_batch(ep, batch_entries, batch_n);
for (int bi = 0; bi < batch_n; bi++) {
free(batch_entries[bi]->raw_payload);
free(batch_entries[bi]);
}
batch_n = 0;
batch_bytes = 0;
}
close(fd);
return -EIO;
}
entry->raw_payload = locked_pbuf;
entry->raw_payload_len = hdr.payload_len;
}
batch_entries[batch_n] = entry;
batch_n++;
batch_bytes += sizeof(wal_entry_hdr_t) + hdr.payload_len;
/* Flush when batch is full (by count or accumulated size) */
if (batch_n >= WAL_STREAM_BATCH_ENTRIES ||
batch_bytes >= WAL_STREAM_BATCH_BYTES) {
send_wal_batch(ep, batch_entries, batch_n);
for (int bi = 0; bi < batch_n; bi++) {
free(batch_entries[bi]->raw_payload);
free(batch_entries[bi]);
}
batch_n = 0;
batch_bytes = 0;
}
}
/* Advance the prev tracker for EVERY scanned entry — including those
* at/below standby_lsn that are not sent — so the first streamed
* entry's prev tuple references the entry the standby already holds
* (prev_log_index == standby_lsn). (Skips on the OOM path above set
* it inline before continuing.)
*
* bug-1028: source the term from the AUTHORITATIVE per-LSN term index
* (persist_entry_term), NOT the raw on-disk hdr.term. The standby's
* logOk and the library send path both normalise via persist_entry_term
* (legacy-0 → WAL_LEGACY_TERM); sending raw hdr.term (0 for a legacy
* entry) makes prev_log_term diverge from the standby's term → logOk
* rejects the catch-up forever. p may be NULL only in count-only tests. */
prev_streamed_lsn = hdr.lsn;
prev_streamed_term = p ? persist_entry_term(p, hdr.lsn) : hdr.term;
off += (uint32_t)(sizeof(hdr) + hdr.payload_len);
}
close(fd);
}
/* Flush remaining partial batch (Req 5.4: send immediately, don't wait) */
if (batch_n > 0) {
send_wal_batch(ep, batch_entries, batch_n);
for (int bi = 0; bi < batch_n; bi++) {
free(batch_entries[bi]->raw_payload);
free(batch_entries[bi]);
}
}
INFO("ha_stream_wal_to_standby: streamed WAL [%" PRIu64 "+1..%" PRIu64
"] to standby", standby_lsn, committed_lsn);
return 0;
}
/* ── ACK-driven catch-up continuation (ha-convergence-recovery task 4.2) ──────
*
* Design D6 (Req 5.1, 5.2, 7.1). The walAck branch of
* repl_ack_decode_and_dispatch (ha.c, iomgr) calls ha_walack_on_ack after
* repl_on_ack; the unit test drives the same logic through
* ha_walack_continuation_for_test. Streaming reads WAL files, so it must never
* run inline on iomgr — the decision schedules a one-shot delaymgr task that
* does the streaming and clears the per-standby single-flight flag.
*
* This lives in config_repl.c (per design) alongside ha_stream_wal_to_standby:
* the continuation wraps the existing streaming primitive with no change to its
* internals.
* ──────────────────────────────────────────────────────────────────────────── */
/* repl_ep_still_registered: is `ep` still an active standby slot? The ep may
* have RDHUPed between schedule and run; on its disappearance the continuation
* is a no-op (design Error Handling: clear the flag and return, no error). */
static int repl_ep_still_registered(replicator_t *r, repl_ep_t ep)
{
int found = 0;
mylock_lock(&r->lock);
for (int i = 0; i < MAX_STANDBYS; i++) {
if (r->standbys[i].ep == ep) {
found = 1;
break;
}
}
mylock_unlock(&r->lock);
return found;
}
/* re-derive the replication target via the SAME pure function
* repl_ack_decode_and_dispatch (ha.c) uses, so this continuation path and
* the heartbeat decision path can never disagree on what a standby must
* catch up to (bug-1478: this function used to duplicate the raw
* max(committed_lsn, last_written) formula inline, independently of the
* ha.c call site the fix landed in — a local-only, never-replicated
* WAL_CHECKPOINT trailing an already-advancing committed_lsn permanently
* wedged catch-up here as well, via the exact same mechanism). */
static uint64_t catchup_derive_target(cluster_t *c)
{
uint64_t committed = atomic_load_explicit(&c->persist.committed_lsn,
memory_order_acquire);
pthread_mutex_lock(&c->persist.lock);
uint64_t last_written = c->persist.current_lsn > 0
? c->persist.current_lsn - 1
: 0;
pthread_mutex_unlock(&c->persist.lock);
return ha_compute_repl_target_lsn(committed, last_written);
}
/* Invoke the catch-up streaming path for [standby_lsn+1, target]. Mirrors the
* HA_REPL_STREAM_WAL dispatch in ha.c: under XCE_TESTING the
* ha_repl_stream_wal_hook seam intercepts so unit tests can count the stream
* without a real endpoint or WAL directory; otherwise ha_stream_wal_to_standby
* runs unchanged. */
static void catchup_stream_range(cluster_t *c, repl_ep_t ep,
uint64_t standby_lsn, uint64_t target)
{
#ifdef XCE_TESTING
if (ha_repl_stream_wal_hook != NULL) {
const char *wal_path = (c && c->persist.wal_path[0])
? c->persist.wal_path : "";
ha_repl_stream_wal_hook(ep, wal_path, standby_lsn, target);
return;
}
#endif
if (c && c->persist.wal_path[0]) {
/* bug-1067: pass the leader's CURRENT term so historical catch-up entries
* carry the AE header term, not their (lower) stored term. */
uint64_t leader_term =
atomic_load_explicit(&c->ha.term, memory_order_acquire);
ha_stream_wal_to_standby((endpoint_t *)ep, c->persist.wal_path,
&c->persist, leader_term, standby_lsn, target);
}
}
/* ha_catchup_continuation_ctx_t — heap context for the one-shot delaymgr task.
* Carries the standby slot identity (peer_idx — stable schedule→run, indexes the
* single-flight flag) and the ACKed LSN; target is re-derived at run time so the
* stream extends to entries appended meanwhile (Req 5.1). */
typedef struct {
cluster_t *c;
repl_ep_t ep;
int peer_idx;
uint64_t acked_lsn;
} ha_catchup_continuation_ctx_t;
/*
* g_drive_async_catchup_offload / ha_set_catchup_offload_for_test /
* ha_catchup_offload_enabled (xce-audit-invariants controller pass,
* thread-role violation 2 fix, 2026-07-18):
*
* Gates whether ha_catchup_continuation_fn hands the WAL-streaming body off
* to the slow lane (ON — closes the thread-role violation) or runs it
* INLINE on the calling (delaymgr) thread (OFF, the PRODUCTION DEFAULT for
* now — see below). Catch-up can stream a variable, potentially large
* amount of WAL data, so leaving it inline on delaymgr — the SAME single
* thread that drives heartbeats and election timeouts cluster-wide — is an
* unbounded block and a real spurious-election risk, not just a theoretical
* one (ha-async-io.md thread-role contract). This is the SAME established
* slow-lane pattern already used for snapshot send, config-push, and peer
* connect in ha.c — see ha_set_demotion_offload_for_test (ha.h) for why NOT
* the vtask pool that deadlocked controller_unit_test on 2026-07-10
* (bug-1429) when tried here.
*
* OFF by default: ha_walack_on_ack / ha_catchup_continuation_fn is exercised
* by test_convergence_verification.c's pre-existing catch-up fixtures, which
* (like most controller_unit_test HA fixtures) never call slowlane_init.
* Flipping this flag ON by default was tried and reproduced a REAL hang —
* the streaming work is silently submitted to an uninitialized
* c->slowlane (queued, never run — no live workers), and downstream code
* waiting on that now-never-happening completion hangs. Matches the
* ORIGINAL rollout discipline of the snapshot/config-push/reconnect
* offloads (all shipped OFF-by-default, flipped ON only in a LATER,
* dedicated pass — raft-drive-async task 3.4 — once the full test suite was
* audited). This change lands the mechanism, correct and covered by its
* own targeted tests below; flipping the cluster-wide default to ON is the
* same kind of follow-up audit, out of scope for this fix (bug-1451,
* buglog.json). Compiled in ALL builds; the setter is the test seam to
* flip RED<->GREEN.
*/
static _Atomic int g_drive_async_catchup_offload = 0;
void ha_set_catchup_offload_for_test(int on)
{
atomic_store_explicit(&g_drive_async_catchup_offload, on ? 1 : 0,
memory_order_release);
}
int ha_catchup_offload_enabled(void)
{
return atomic_load_explicit(&g_drive_async_catchup_offload,
memory_order_acquire);
}
/*
* catchup_lane_fn — slow-lane worker entry (and the legacy inline-path body,
* offload flag OFF). Re-validate the endpoint against the registry,
* re-derive target, stream [acked_lsn+1, target], and clear catchup_inflight
* on EVERY exit path (success, ep-gone, nothing-to-stream) so a stuck flag
* never blocks future continuations (Req 5.2). Frees its own ctx
* (body-frees-own-ctx pattern, mirroring ha_demotion_convergence_do_run).
*
* The blocking work itself — catchup_stream_range -> ha_stream_wal_to_standby
* (WAL file reads + socket writes to a standby) — is unchanged from before
* the offload; only WHERE it runs changed (see ha_catchup_continuation_fn).
*/
static void catchup_lane_fn(void *arg)
{
ha_catchup_continuation_ctx_t *ctx = (ha_catchup_continuation_ctx_t *)arg;
if (ctx == NULL)
return;
cluster_t *c = ctx->c;
repl_ep_t ep = ctx->ep;
int peer_idx = ctx->peer_idx;
uint64_t acked = ctx->acked_lsn;
/* Endpoint may have RDHUPed between schedule and run — no-op if gone. */
if (!repl_ep_still_registered(&c->persist.repl, ep)) {
INFO("HA catch-up continuation: standby[%d] endpoint gone before run "
"— skipping (heartbeat fallback re-evaluates)", peer_idx);
goto clear_and_done;
}
uint64_t target = catchup_derive_target(c);
if (acked < target) {
INFO("HA catch-up continuation: standby[%d] acked=%" PRIu64
" < target=%" PRIu64 " → streaming [%" PRIu64 "+1..%" PRIu64 "]",
peer_idx, acked, target, acked, target);
catchup_stream_range(c, ep, acked, target);
}
/* If acked >= target the standby reached parity meanwhile — nothing to
* stream; parity re-inclusion is owned by the decision path, not here. */
clear_and_done:
/* Req 5.2: release the single-flight slot so the next lagging ACK schedules
* again. release order so the decision path's acquire load observes it. */
if (peer_idx >= 0 && peer_idx < MAX_STANDBYS)
atomic_store_explicit(&c->ha.catchup_inflight[peer_idx], 0,
memory_order_release);
free(ctx);
}
/*
* ha_catchup_continuation_fn — delaymgr task body (the FIRST hop, off the
* iomgr walAck thread — see ha_schedule_catchup_continuation below). With
* the offload flag ON, this function does only fast, non-blocking admin
* work (slowlane_submit) and returns immediately — the SECOND hop, off the
* delaymgr thread itself, onto a slow-lane worker (catchup_lane_fn), which
* is where catchup_stream_range's blocking WAL reads + socket writes
* actually run. With the flag OFF (the current default — see
* g_drive_async_catchup_offload above), runs the legacy inline blocking
* path.
*
* c->ha.catchup_inflight[peer_idx] (set by ha_walack_on_ack BEFORE this task
* is even scheduled) remains the authoritative per-standby single-flight
* guard (Req 5.2); the lane's own per-key dedup (key = peer_idx+1) is a
* redundant backstop — same relationship ha_connect_to_peer's reconnect
* offload has with reconnect_inflight[] (ha.c).
*
* History: a VTASK_LIVE_BLOCK vtask-pool dispatch was attempted here on
* 2026-07-10 to get this off the delaymgr thread and deadlocked
* controller_unit_test — reverted (bug-1429). The slow lane (this fix) is a
* different, already-proven mechanism (raft-drive-async's snapshot/
* config-push/reconnect offloads) and does not share that failure mode: its
* worker threads start unconditionally in cluster_init_phase2, and
* slowlane_submit never blocks the caller even when a lightweight unit-test
* fixture never calls slowlane_init (it just queues work that will not run,
* rather than blocking waiting for a worker).
*/
int ha_catchup_continuation_fn(void *arg)
{
ha_catchup_continuation_ctx_t *ctx = (ha_catchup_continuation_ctx_t *)arg;
if (ctx == NULL || ctx->c == NULL) {
free(ctx);
return -EINVAL;
}
cluster_t *c = ctx->c;
int peer_idx = ctx->peer_idx;
if (!ha_catchup_offload_enabled()) {
/* Legacy inline path (flag OFF) — blocks the calling (delaymgr)
* thread. Retained solely so the RED test can show the pre-change
* stall; production runs with the flag ON. */
catchup_lane_fn(ctx);
return 0;
}
uint64_t key = (uint64_t)(peer_idx + 1);
int srv = slowlane_submit(&c->slowlane, key, catchup_lane_fn, ctx);
if (srv == -EEXIST) {
/* Lane still holds this peer's dedup key from a PRIOR continuation
* that already cleared catchup_inflight[peer_idx] (its last action)
* but whose key the lane has not yet removed. Release the guard so
* the next WAL ack retries cleanly (mirrors ha_connect_to_peer's
* -EEXIST handling in ha.c). */
if (peer_idx >= 0 && peer_idx < MAX_STANDBYS)
atomic_store_explicit(&c->ha.catchup_inflight[peer_idx], 0,
memory_order_release);
free(ctx);
return 0;
}
if (srv < 0 && srv != -EAGAIN) {
/* Hard submit failure (alloc / lane stopping). -EAGAIN is NOT a
* failure (still queued, will run). Only on a real failure release
* the guard and drop — the next WAL ack schedules a fresh
* continuation (never self-terminate; this event recurs on every
* ack, unlike the one-shot demotion-convergence task). */
WARN("HA catch-up continuation: slowlane_submit failed for "
"standby[%d]: %d — retry next ack", peer_idx, srv);
if (peer_idx >= 0 && peer_idx < MAX_STANDBYS)
atomic_store_explicit(&c->ha.catchup_inflight[peer_idx], 0,
memory_order_release);
free(ctx);
return 0;
}
/* 0 or -EAGAIN: enqueued exactly once; catchup_lane_fn clears the flag
* and frees ctx on completion. */
return 0;
}
/*
* ha_schedule_catchup_continuation — enqueue one one-shot delaymgr task
* running ha_catchup_continuation_fn directly. The single-flight flag is set
* by the caller (ha_walack_on_ack) before this call; on a scheduling failure
* this clears it so the slot does not stay stuck. Returns 0 on success,
* negative on failure.
*/
static int ha_schedule_catchup_continuation(cluster_t *c, repl_ep_t ep,
int peer_idx, uint64_t acked_lsn)
{
ha_catchup_continuation_ctx_t *ctx = calloc(1, sizeof(*ctx));
if (ctx == NULL) {
ERROR("HA catch-up continuation: ctx alloc failed for standby[%d]",
peer_idx);
atomic_store_explicit(&c->ha.catchup_inflight[peer_idx], 0,
memory_order_release);
return -ENOMEM;
}
ctx->c = c;
ctx->ep = ep;
ctx->peer_idx = peer_idx;
ctx->acked_lsn = acked_lsn;
/* One-shot (runcount=1), minimal delay: streaming reads WAL files and must
* run off the iomgr walAck thread. delay_task_ms only enqueues (ordq_put);
* the delaymgr worker runs ha_catchup_continuation_fn itself. In
* XCE_TESTING the fixture drives the queued task synchronously via its
* stored entry/args. */
int rc = delay_task_ms(&c->delaymgr, 1, 1,
(delay_task_entry_fn_t)ha_catchup_continuation_fn,
ctx);
if (rc < 0) {
ERROR("HA catch-up continuation: delay_task_ms failed for standby[%d]:"
" %d", peer_idx, rc);
free(ctx);
atomic_store_explicit(&c->ha.catchup_inflight[peer_idx], 0,
memory_order_release);
return rc;
}
return 0;
}
int ha_walack_on_ack(cluster_t *c, void *ep, int peer_idx, uint64_t acked_lsn)
{
if (c == NULL || ep == NULL)
return -EINVAL;
if (peer_idx < 0 || peer_idx >= MAX_STANDBYS)
return -EINVAL;
uint64_t target = catchup_derive_target(c);
/* Parity (Req 7.1): the standby reached the boundary — re-include it in live
* replication via the existing-but-untriggered re-inclusion call. No
* continuation; ha-commit-engine wiring propagates commit_on_membership_change. */
if (acked_lsn >= target) {
repl_complete_catchup(&c->persist.repl, (repl_ep_t)ep);
return 0;
}
/* Behind target (Req 5.1/5.2): single-flight — claim the per-standby slot
* 0→1; if already in flight, schedule nothing. */
int expected = 0;
if (!atomic_compare_exchange_strong_explicit(
&c->ha.catchup_inflight[peer_idx], &expected, 1,
memory_order_acq_rel, memory_order_acquire)) {
/* A continuation is already in flight for this standby (Req 5.2). */
return 0;
}
return ha_schedule_catchup_continuation(c, (repl_ep_t)ep, peer_idx,
acked_lsn);
}
/* ── Snapshot chunk transfer ───────────────────────────────────────────── */
#ifdef XCE_TESTING
int (*xce_snapshot_chunk_hook)(endpoint_t *ep, xce_request_t *req) = NULL;
#endif
#define SNAPSHOT_CHUNK_SIZE 65536
int ha_send_snapshot_to_standby(endpoint_t *ep, const char *snap_dir,
uint64_t checkpoint_lsn,
uint64_t checkpoint_term)
{
if (!snap_dir)
return -EINVAL;
char snap_path[4096];
snprintf(snap_path, sizeof(snap_path),
"%s/snap-%016" PRIx64 ".capnp", snap_dir, checkpoint_lsn);
/* Stat the snapshot file to get total_size */
struct stat st;
if (stat(snap_path, &st) != 0)
return -errno;
uint64_t total_size = (uint64_t)st.st_size;
/* Req 7.3: log snapshot transfer start with checkpoint_lsn and size */
INFO("ha_send_snapshot_to_standby: snapshot transfer start "
"checkpoint_lsn=%" PRIu64 " size=%" PRIu64,
checkpoint_lsn, total_size);
FILE *f = fopen(snap_path, "rb");
if (!f)
return -errno;
uint8_t *buf = malloc(SNAPSHOT_CHUNK_SIZE);
if (!buf) {
fclose(f);
return -ENOMEM;
}
int rc = 0;
uint64_t offset = 0;
uint32_t chunk_index = 0;
/* Chunk the snapshot file and enqueue each chunk with its file bytes.
* Payload travels as capnp Data via xce_snapshot_chunk_req_raw (task 7.2)
* — the generated encode path cannot carry raw bytes (xce_data_t
* limitation, same as replConfigPush). */
while (offset < total_size) {
size_t chunk_len = (total_size - offset > SNAPSHOT_CHUNK_SIZE)
? SNAPSHOT_CHUNK_SIZE
: (size_t)(total_size - offset);
if (fread(buf, 1, chunk_len, f) != chunk_len) {
rc = -EIO;
break;
}
/* bug-1389 triage: crc32 of the exact bytes this chunk is about to
* send, taken right after the local file read and before encode —
* correlate against ha_handle_snapshot_chunk's receive-side crc32
* log (ha.c) by (snapshot_lsn, chunk_index) to localize whether a
* mismatch is introduced before send, in the wire transport, or
* during receive-side assembly. */
TRACE("ha_send_snapshot_to_standby: chunk crc32 "
"checkpoint_lsn=%" PRIu64 " chunk_index=%u offset=%" PRIu64
" len=%zu crc32=%08x",
checkpoint_lsn, chunk_index, offset, chunk_len,
(unsigned)crc32(0L, (const Bytef *)buf, (uInt)chunk_len));
int is_last = (offset + chunk_len >= total_size);
snapshot_transfer_t chunk;
memset(&chunk, 0, sizeof(chunk));
chunk.snapshot_lsn = checkpoint_lsn;
chunk.chunk_index = chunk_index;
chunk.chunk_offset = offset;
chunk.total_size = total_size;
chunk.is_last = is_last;
/* raft-ha-core-safety task 3.2: stamp every chunk with the
* checkpoint's own term so the standby can reconcile its per-LSN
* term index at the boundary regardless of which chunk it last
* processes (is_last framing already exists per-chunk; cheap to
* repeat). */
chunk.checkpoint_term = checkpoint_term;
chunk.raw_payload = buf; /* borrowed for hook/encode only */
chunk.raw_payload_len = chunk_len;
/* bug-1386: a failed chunk send (e.g. -ECONNRESET while the standby
* endpoint is mid-teardown/reconnect) MUST stop the transfer and be
* propagated. Previously this return value was discarded, so the
* loop kept "sending" past a dead endpoint and the function still
* reported success — the standby silently received a truncated
* snap.part that never gets an is_last chunk. */
int send_rc = 0;
#ifdef XCE_TESTING
if (xce_snapshot_chunk_hook) {
xce_request_t req;
memset(&req, 0, sizeof(req));
req.opcode = OPCode_snapshotTransfer;
req.body_which = Request_body_snapshottransfermsg;
req.body.snapshottransfermsg = &chunk;
send_rc = xce_snapshot_chunk_hook(ep, &req);
} else
#endif
if (ep)
send_rc = xce_snapshot_chunk_req_raw(ep, &chunk, buf, chunk_len);
if (send_rc != 0) {
rc = send_rc;
break;
}
offset += chunk_len;
chunk_index++;
}
free(buf);
fclose(f);
/* Req 7.3: log snapshot transfer complete (on success only) */
if (rc == 0)
INFO("ha_send_snapshot_to_standby: snapshot transfer complete "
"checkpoint_lsn=%" PRIu64 " size=%" PRIu64,
checkpoint_lsn, total_size);
else
WARN("ha_send_snapshot_to_standby: snapshot transfer failed "
"checkpoint_lsn=%" PRIu64 " size=%" PRIu64 " sent=%" PRIu64
" chunk_index=%u rc=%d",
checkpoint_lsn, total_size, offset, chunk_index, rc);
return rc;
}
uint64_t ha_compute_oldest_wal_lsn(const char *wal_path)
{
DIR *d = opendir(wal_path);
if (!d)
return 0;
uint64_t oldest = UINT64_MAX;
struct dirent *ent;
while ((ent = readdir(d)) != NULL) {
uint64_t seg_lsn = 0;
if (sscanf(ent->d_name, "seg-%" SCNx64 ".wal", &seg_lsn) != 1)
continue;
if (seg_lsn < oldest)
oldest = seg_lsn;
}
closedir(d);
return (oldest == UINT64_MAX) ? 0 : oldest;
}