From 2c375cf4764d9408c5831c7fccb12fab078b0f35 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Wed, 13 May 2026 14:15:30 +0000 Subject: [PATCH] feat: add config replication fields to myconf_t, cluster_t, ha_state_t, persist_t; create byteorder.h --- src/controller/ha.h | 53 +++++++++++++++++++++++++++++++++++ src/include/byteorder.h | 18 ++++++++++++ src/include/myconf.h | 17 +++++++++++ src/include/persist.h | 5 +++- src/include/xce/cluster_def.h | 4 +++ 5 files changed, 96 insertions(+), 1 deletion(-) create mode 100644 src/include/byteorder.h diff --git a/src/controller/ha.h b/src/controller/ha.h index eacfff6..07cba6a 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -18,16 +18,45 @@ * The watchdog (P1-3) will include netio.h directly where it needs endpoint_t. */ +#include #include #include #include #include +#include #include "config.h" /* XCE_MAXNAME_LEN, MAX_STANDBYS, etc. */ +#include "zarray.h" /* Forward declaration — avoids circular include with cluster_def.h */ typedef struct cluster_s cluster_t; +/* ── Inline PRNG for election jitter ───────────────────────── */ + +static inline uint64_t xorshift64_next(uint64_t *state) +{ + uint64_t x = *state; + x ^= x << 13; + x ^= x >> 7; + x ^= x << 17; + *state = x; + return x; +} + +static inline void ha_seed_jitter(uint64_t *seed, int my_priority) +{ + int fd = open("/dev/urandom", O_RDONLY); + if (fd >= 0) { + if (read(fd, seed, 8) == 8 && *seed != 0) { close(fd); return; } + close(fd); + } + struct timespec ts; + clock_gettime(CLOCK_MONOTONIC, &ts); + *seed = (uint64_t)ts.tv_nsec + ^ ((uint64_t)my_priority << 56) + ^ ((uint64_t)getpid() << 16); +} + /* ── Role enum ─────────────────────────────────────────────── */ typedef enum { CTRL_PRIMARY = 0, @@ -42,6 +71,8 @@ typedef enum { #define HA_DEFAULT_ELECTION_TIMEOUT_MS 2000 #define HA_DEFAULT_PROBE_CONNECT_TIMEOUT_MS 500 #define HA_DEFAULT_MAX_PROMOTION_RETRIES 5 +#define HA_DEFAULT_ELECTION_JITTER_MAX_MS 200 +#define HA_DEFAULT_ELECTION_PRIORITY_STEP_MS 500 typedef struct { uint32_t heartbeat_interval_ms; /* Primary heartbeat send interval (min 2000 ms) */ @@ -49,8 +80,22 @@ typedef struct { uint32_t election_timeout_ms; /* Election probe shared deadline (min 500 ms) */ uint32_t probe_connect_timeout_ms; /* Per-probe TCP connect timeout */ uint32_t max_promotion_retries; /* Max consecutive failed promotion attempts */ + uint32_t election_jitter_max_ms; /* random jitter added to election timeout (default 200) */ + uint32_t election_priority_step_ms; /* extra ms per priority level (default 500) */ } ha_config_t; +/* Forward declaration for election probe context */ +struct xce_candidate_s; + +typedef struct { + struct xce_candidate_s *candidate; /* peer to probe */ + void *ssl_ctx; /* SSL_CTX* (void* to avoid openssl header) */ + uint32_t timeout_ms; + atomic_int *election_done; + int got_ack; + pthread_t thread; +} probe_ctx_t; + /* ── HA runtime state (embedded in cluster_t as `ha_state_t ha` in P1-5) ── */ typedef struct { _Atomic uint64_t term; /* Election term — fencing token */ @@ -71,6 +116,9 @@ typedef struct { /* Watchdog thread — started in P1-3 */ _Atomic int watchdog_stop; pthread_t watchdog_thread; + void *ssl_ctx; /* SSL_CTX* cast; avoids pulling openssl headers here */ + uint64_t jitter_seed; /* xorshift64 PRNG state for election jitter */ + zarray_t *standby_endpoints; /* connected standby controller endpoints for config push */ } ha_state_t; /* ── Public API ────────────────────────────────────────────── */ @@ -166,6 +214,11 @@ int ha_should_redirect(const ha_state_t *ha); */ int ha_no_primary_known(const ha_state_t *ha); +int ha_is_primary(const ha_state_t *ha); +void ha_check_priority_stepdown(ha_state_t *ha, int sender_priority, uint64_t sender_term); +void start_heartbeat_listener(cluster_t *c); +void stop_heartbeat_listener(cluster_t *c); + /* ── P3-6: Test-only failover shim ────────────────────────────── */ #ifdef XCE_TESTING diff --git a/src/include/byteorder.h b/src/include/byteorder.h new file mode 100644 index 0000000..1010afc --- /dev/null +++ b/src/include/byteorder.h @@ -0,0 +1,18 @@ +#pragma once +/* Portable be64toh() for non-glibc or pre-2.25 targets. */ +#if defined(__linux__) +# include +#elif defined(__APPLE__) +# include +# include +# define be64toh(x) OSSwapBigToHostInt64(x) +#else +# include +static inline uint64_t be64toh(uint64_t x) { +# if __BYTE_ORDER__ == __ORDER_LITTLE_ENDIAN__ + return __builtin_bswap64(x); +# else + return x; +# endif +} +#endif diff --git a/src/include/myconf.h b/src/include/myconf.h index fcfe762..6d21379 100644 --- a/src/include/myconf.h +++ b/src/include/myconf.h @@ -4,11 +4,17 @@ #define MAX_MASTER_CANDIDATES (64) +#include +#include + #include #include "clusterconfig.h" #include "xce/cluster.h" +/* Forward declaration — avoids circular include with cluster_def.h */ +typedef struct cluster_s cluster_t; + struct logging { char* logdir; char* loglevel; @@ -22,9 +28,20 @@ typedef struct { char* workdir; /* working directory for WAL/snapshot storage */ char* name; /* controller name (hostname by default) */ xce_cluster_config_t peers_cfg; /* HA peer list parsed from --peers; count==0 = single-node */ + uint64_t config_version_tag; /* first 8 bytes of SHA-256, host order */ + uint8_t config_sha256[32]; /* full SHA-256 of conffile bytes */ + int conf_keep_n; /* versioned file retention count (default 2) */ + uint64_t config_generation; /* monotonic audit counter */ + time_t conf_stale_timeout; /* seconds before saved config flagged stale (default 86400) */ + int config_stale; /* 1 when saved config older than conf_stale_timeout */ } myconf_t; extern cluster_config_t* get_default_conf(void*); extern int add_config(void*, char*, char*); +void myconf_fini(myconf_t *myconf); + +int load_primary_config(cluster_t *c); +int fetch_config_from_primary(cluster_t *c); + #endif diff --git a/src/include/persist.h b/src/include/persist.h index 1ba0a8c..e850696 100644 --- a/src/include/persist.h +++ b/src/include/persist.h @@ -24,6 +24,7 @@ typedef struct { uint32_t finished_retain_s; /* retain finished jobs N seconds (default 3600) */ uint32_t worker_retain_s; /* retain unavailable workers N seconds (default 86400) */ uint32_t sync_timeout_ms; /* sync replication ACK timeout ms (default 50) */ + uint32_t max_wal_records; /* checkpoint trigger: records since last snap (default 10000) */ } persist_config_t; static inline persist_config_t persist_config_defaults(void) @@ -37,6 +38,7 @@ static inline persist_config_t persist_config_defaults(void) c.finished_retain_s = DEFAULT_FINISHED_RETAIN_S; c.worker_retain_s = DEFAULT_WORKER_RETAIN_S; c.sync_timeout_ms = DEFAULT_SYNC_TIMEOUT_MS; + c.max_wal_records = 10000; return c; } @@ -78,12 +80,13 @@ struct persist_s { /* Concurrency */ pthread_mutex_t lock; + _Atomic uint32_t wal_record_count; /* records appended since last checkpoint */ + _Atomic int checkpoint_in_progress; /* CAS guard: 0=idle, 1=scheduled/running */ atomic_int fsync_stop; atomic_int fsync_thread_running; pthread_t fsync_thread; /* Checkpoint (stubs for now — wired in Task 10) */ - atomic_int checkpoint_in_progress; atomic_int checkpoint_requested; pthread_cond_t checkpoint_done; pthread_mutex_t checkpoint_mutex; diff --git a/src/include/xce/cluster_def.h b/src/include/xce/cluster_def.h index ed7ae3d..6756263 100644 --- a/src/include/xce/cluster_def.h +++ b/src/include/xce/cluster_def.h @@ -120,6 +120,10 @@ typedef struct cluster_s { int running; persist_t persist; /* Persistence module — P0-F */ ha_state_t ha; /* HA role management — P1-5 */ + _Atomic int restart_requested; /* set by OP_upgradeConfig handler */ + pthread_t heartbeat_listener_tid; /* pre-Phase-2 SSL listener thread */ + _Atomic int heartbeat_listener_running; + _Atomic int lost_and_found_job_count; /* incremented during persist_recover() */ } cluster_t; #endif /* XCE_CLUSTER_DEF_H */