1963 lines
68 KiB
C
1963 lines
68 KiB
C
#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 "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 */
|
||
|
||
/* 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: <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);
|
||
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;
|
||
}
|
||
|
||
/* Save to disk and reload */
|
||
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;
|
||
}
|
||
|
||
rc = load_primary_config(c);
|
||
if (rc != 0) {
|
||
WARN("handle_repl_config_push: load_primary_config failed: %d", rc);
|
||
return rc;
|
||
}
|
||
|
||
INFO("handle_repl_config_push: config saved and loaded (tag=%016" PRIx64 ")",
|
||
c->myconf->config_version_tag);
|
||
|
||
/* 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
|
||
|
||
static void *push_thread(void *arg)
|
||
{
|
||
push_ctx_t *ctx = (push_ctx_t *)arg;
|
||
|
||
int rc = 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);
|
||
|
||
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).
|
||
* 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;
|
||
}
|
||
pthread_mutex_unlock(&c->ha.lock);
|
||
|
||
pthread_t *threads = malloc((size_t)n * sizeof(pthread_t));
|
||
if (!threads) {
|
||
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);
|
||
}
|
||
|
||
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). */
|
||
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. Push to standbys — the majority-ACK barrier result is
|
||
* authoritative for the upgrade outcome (barrier-rollout Req 4.1/4.2):
|
||
* non-zero means the new tag never reached quorum, so the requester
|
||
* gets an explicit error and the cluster stays on the old committed
|
||
* tag (no success ACK, no restart — Req 4.3/4.4). Disk save and
|
||
* generation bump have already happened (pre-existing sequence);
|
||
* convergence of a half-pushed config remains the OP_replConfig
|
||
* anti-entropy behavior. */
|
||
int acked = 0;
|
||
rc = push_config_to_standbys(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 push failed to reach majority quorum"
|
||
" -- 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);
|
||
|
||
/* 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;
|
||
#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;
|
||
}
|
||
|
||
int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path,
|
||
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;
|
||
|
||
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;
|
||
|
||
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;
|
||
}
|
||
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;
|
||
|
||
if (hdr.payload_len > 0) {
|
||
uint8_t *pbuf = malloc(hdr.payload_len);
|
||
if (pbuf) {
|
||
ssize_t pr = pread(fd, pbuf, hdr.payload_len,
|
||
(off_t)(off + sizeof(hdr)));
|
||
if (pr == (ssize_t)hdr.payload_len) {
|
||
entry->raw_payload = pbuf;
|
||
entry->raw_payload_len = hdr.payload_len;
|
||
} else {
|
||
free(pbuf);
|
||
}
|
||
}
|
||
}
|
||
|
||
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;
|
||
}
|
||
}
|
||
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])
|
||
ha_stream_wal_to_standby((endpoint_t *)ep, c->persist.wal_path,
|
||
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)
|
||
{
|
||
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;
|
||
}
|
||
|
||
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;
|
||
chunk.raw_payload = buf; /* borrowed for hook/encode only */
|
||
chunk.raw_payload_len = chunk_len;
|
||
|
||
#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;
|
||
xce_snapshot_chunk_hook(ep, &req);
|
||
} else
|
||
#endif
|
||
if (ep)
|
||
xce_snapshot_chunk_req_raw(ep, &chunk, buf, chunk_len);
|
||
|
||
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);
|
||
|
||
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;
|
||
}
|