diff --git a/src/controller/ha.c b/src/controller/ha.c index 2159430..788597e 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -2,18 +2,47 @@ #include "ha.h" #include +#include #include #include #include #include #include #include +#include +#include /* Circular-include safety: cluster_def.h includes ha.h (include-guarded), * and ha.h only forward-declares cluster_t — no cycle. */ #include "xce/cluster_def.h" #include "persist.h" +/* ── PRNG seeding ───────────────────────────────────────────── */ + +void ha_seed_jitter(uint64_t *seed, int my_priority) +{ + uint8_t buf[8] = {0}; + int fd = open("/dev/urandom", O_RDONLY); + if (fd >= 0) { + ssize_t n = 0; + while (n < 8) { + ssize_t r = read(fd, buf + n, (size_t)(8 - n)); + if (r <= 0) break; + n += r; + } + close(fd); + if (n == 8) { + memcpy(seed, buf, 8); + if (*seed != 0) return; + } + } + struct timespec ts; + clock_gettime(CLOCK_MONOTONIC, &ts); + *seed = (uint64_t)ts.tv_nsec + ^ ((uint64_t)my_priority << 56) + ^ ((uint64_t)getpid() << 16); +} + int ha_init(ha_state_t *ha, const ha_config_t *cfg, int my_priority) { if (!ha || !cfg) diff --git a/src/controller/ha.h b/src/controller/ha.h index 30b6e2c..19c4cd7 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -18,15 +18,13 @@ * The watchdog (P1-3) will include netio.h directly where it needs endpoint_t. */ -#include #include #include #include -#include #include -#include -#include "config.h" /* XCE_MAXNAME_LEN, MAX_STANDBYS, etc. */ +#include "config.h" /* XCE_MAXNAME_LEN, MAX_STANDBYS, etc. */ +#include "xce/cluster.h" /* xce_candidate_t */ #include "zarray.h" /* Forward declaration — avoids circular include with cluster_def.h */ @@ -44,30 +42,12 @@ static inline uint64_t xorshift64_next(uint64_t *state) return x; } -static inline void ha_seed_jitter(uint64_t *seed, int my_priority) -{ - uint8_t buf[8] = {0}; - int fd = open("/dev/urandom", O_RDONLY); - if (fd >= 0) { - ssize_t n = 0; - while (n < 8) { - ssize_t r = read(fd, buf + n, (size_t)(8 - n)); - if (r <= 0) break; - n += r; - } - close(fd); - if (n == 8) { - memcpy(seed, buf, 8); - if (*seed != 0) return; - } - } - /* Fallback: monotonic clock + priority + pid mixing */ - struct timespec ts; - clock_gettime(CLOCK_MONOTONIC, &ts); - *seed = (uint64_t)ts.tv_nsec - ^ ((uint64_t)my_priority << 56) - ^ ((uint64_t)getpid() << 16); -} +/* + * ha_seed_jitter: seed the xorshift64 PRNG from /dev/urandom (8 bytes). + * Falls back to CLOCK_MONOTONIC ^ priority ^ pid if urandom is unavailable. + * Not inline — contains syscalls (open/read/close/clock_gettime/getpid). + */ +void ha_seed_jitter(uint64_t *seed, int my_priority); /* ── Role enum ─────────────────────────────────────────────── */ typedef enum { @@ -96,11 +76,8 @@ typedef struct { 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 */ + xce_candidate_t *candidate; /* peer to probe */ void *ssl_ctx; /* SSL_CTX* (void* to avoid openssl header) */ uint32_t timeout_ms; atomic_int *election_done; @@ -226,9 +203,32 @@ int ha_should_redirect(const ha_state_t *ha); */ int ha_no_primary_known(const ha_state_t *ha); +/* + * ha_is_primary: returns 1 if ha->role is CTRL_PRIMARY, 0 otherwise. + * Lock-free. Safe to call from any thread. + */ 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); + +/* + * ha_check_priority_stepdown: called from the heartbeat handler when + * term == 0 (simultaneous boot). If sender_priority < ha->my_priority + * (higher priority), steps down the local node from PRIMARY to STANDBY. + * No-op if term > 0 or sender is not higher priority. + */ +void ha_check_priority_stepdown(ha_state_t *ha, int sender_priority, + uint64_t sender_term); + +/* + * start_heartbeat_listener: start a minimal pre-Phase-2 SSL listener + * that responds to OP_heartbeat probes. Called after am_i_primary() + * returns 1. Records tid in c->heartbeat_listener_tid. + */ void start_heartbeat_listener(cluster_t *c); + +/* + * stop_heartbeat_listener: signal and join the heartbeat listener thread. + * Must be called before iomgr_init() to free the listening port (EADDRINUSE). + */ void stop_heartbeat_listener(cluster_t *c); /* ── P3-6: Test-only failover shim ────────────────────────────── */ diff --git a/src/include/config.h b/src/include/config.h index 384dda3..11eeb1a 100644 --- a/src/include/config.h +++ b/src/include/config.h @@ -92,5 +92,6 @@ typedef uint64_t jobid_t; #define DEFAULT_WORKER_RETAIN_S 86400 #define DEFAULT_SYNC_TIMEOUT_MS 50 #define DEFAULT_HB_INTERVAL_MS 3000 +#define DEFAULT_MAX_WAL_RECORDS 10000 #endif diff --git a/src/include/myconf.h b/src/include/myconf.h index 6d21379..4e45e03 100644 --- a/src/include/myconf.h +++ b/src/include/myconf.h @@ -33,7 +33,7 @@ typedef struct { 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 */ + int conf_stale; /* 1 when saved config older than conf_stale_timeout */ } myconf_t; extern cluster_config_t* get_default_conf(void*); diff --git a/src/include/persist.h b/src/include/persist.h index e850696..d62dc5a 100644 --- a/src/include/persist.h +++ b/src/include/persist.h @@ -38,7 +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; + c.max_wal_records = DEFAULT_MAX_WAL_RECORDS; return c; } @@ -80,8 +80,8 @@ 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_uint 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;