#define _GNU_SOURCE #include "config_repl.h" #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include /* 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 OFF-IOMGR delaymgr task that re-runs config_activate_committed. The * delaymgr worker runs on a thread allowed to block on the disk reads * (scan_latest_committed_marker pread + load_primary_config). 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: the delaymgr task body (runs OFF the iomgr ACK * thread). 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. `arg` is the cluster_t* (passed directly; no heap ctx to free, so the * delaymgr node-killer that frees only the delay_task_t never leaks). */ 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_schedule_activate_retry: enqueue config_activate_retry_fn on the * delaymgr (NON-BLOCKING — delay_task_ms only does ordq_put). 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_fn, 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->port, 0, 9990); 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 ────────────────────────── */ int load_primary_config(cluster_t *c) { myconf_t *myconf = c->myconf; /* Build conf dir path: / */ 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); 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) 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; 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); /* Free any prior decoded config before overwriting the pointer */ if (myconf->cluster_config) { free_ClusterConfig(myconf->cluster_config); free(myconf->cluster_config); myconf->cluster_config = NULL; } /* Decode Cap'n Proto directly into myconf->cluster_config */ rc = load_cluster_config_from_file(myconf->conffile, &myconf->cluster_config); if (rc != 0) { ERROR("Primary config '%s' failed Cap'n Proto decode: %d", myconf->conffile, rc); free(bytes); free(myconf->conffile); myconf->conffile = NULL; return rc; } if (myconf->cluster_config == NULL) { ERROR("Primary config '%s' decoded to NULL", myconf->conffile); free(bytes); free(myconf->conffile); myconf->conffile = NULL; return -EINVAL; } free(bytes); fill_default_cluster_config(myconf->cluster_config); /* Check for reserved lost_and_found name */ if (config_contains_reserved_name(myconf->cluster_config)) { ERROR("Config '%s' contains reserved name 'lost_and_found'", myconf->conffile); free_ClusterConfig(myconf->cluster_config); free(myconf->cluster_config); myconf->cluster_config = NULL; free(myconf->conffile); myconf->conffile = NULL; return -EINVAL; } /* Default generation to 1 if not set */ if (myconf->cluster_config->config_generation == 0) myconf->cluster_config->config_generation = 1; myconf->config_generation = myconf->cluster_config->config_generation; /* 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); 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 ──────────────────────── */ /* * 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) * 9. Wait for in-flight checkpoint, then signal restart */ 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; } /* 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); /* NOTICE (bug-857, spec `.kiro/specs/config-upgrade-snapshot/`) — config * upgrade MUST snapshot current state, not re-derive it per node. * Today the upgrade commits the config-change marker, then each member * RESTARTS and re-materializes its post-upgrade state by replaying its own * pre-upgrade WAL under the NEW config (steps 9 + restart recovery). Replay * is config-dependent (`lost_and_found` re-homes entities whose config * entity vanished), so members can DIVERGE in materialized state after an * upgrade. The fix: at the config-change marker's (term,lsn) boundary the * controller must materialize a full ENTITY-BEARING state snapshot, embed * the new config in it, and install that SAME snapshot on every member via * the snapshot_transfer_s transport (install-before-activate, hash-verified) * so post-upgrade state is identical by construction and no member * re-derives it. HARD dependency: `entity-bearing-snapshots` (snapshots are * a 16-byte LSN sentinel today; the entity payload does not exist yet). * Until that lands, the per-node-replay path below is the interim behavior. */ /* 9. Wait for in-flight checkpoint (up to 30 s), then signal restart */ struct timespec ckpt_start, ckpt_now; clock_gettime(CLOCK_MONOTONIC, &ckpt_start); while (atomic_load(&c->persist.checkpoint_in_progress)) { usleep(10000); clock_gettime(CLOCK_MONOTONIC, &ckpt_now); if (ckpt_now.tv_sec - ckpt_start.tv_sec > 30) { ERROR("Checkpoint did not complete within 30 s" " -- restarting anyway"); break; } } INFO("Config upgraded to tag=%016" PRIx64 " gen=%" PRIu64 " (acked=%d standbys) -- requesting restart", c->myconf->config_version_tag, new_gen, acked); atomic_store(&c->restart_requested, 1); int stop_val = 1; cbuf_put(&c->wbuf, &stop_val); 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; } 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; 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; /* 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 * and do not advance prev_streamed_lsn/prev_streamed_term, so the * NEXT real entry's AppendEntries prev tuple chains to the last * REAL entry before this gap, exactly matching what a standby that * never records a checkpoint marker in its own log expects. * Without this, a standby receiving one gets stuck retrying the * same LSN forever (the receive-side guard correctly refuses it). */ if (hdr.type == WAL_CHECKPOINT) { off += (uint32_t)(sizeof(hdr) + hdr.payload_len); continue; } if (hdr.lsn > standby_lsn) { wal_catchup_entry_t *entry = calloc(1, sizeof(*entry)); if (!entry) { /* 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: read the body with bounded * retry; a non-empty COMMITTED entry must NEVER be shipped * empty (the live LSN-9 send-side payload loss). */ uint8_t *pbuf = NULL; int rrc = stream_read_entry_body( fd, off + (uint32_t)sizeof(hdr), hdr.payload_len, hdr.lsn, entry->term, leader_term, &pbuf); if (rrc != 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, rrc, WAL_STREAM_BODY_READ_MAX_ATTEMPTS); free(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; } close(fd); return -EIO; } entry->raw_payload = 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 exactly as ha.c:2771-2776 / D6: * target = max(committed_lsn, last_written), last_written = current_lsn - 1. */ 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 last_written > committed ? last_written : committed; } /* 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; /* * ha_catchup_continuation_fn — delaymgr task body (runs OFF the iomgr walAck * branch, D6/D1). 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_fn). */ 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; 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); return 0; } /* * ha_schedule_catchup_continuation — enqueue one one-shot delaymgr task running * ha_catchup_continuation_fn. 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 it. 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; }