Rongsong Shen
dbb750ebd3
- ha.c: self-vote in ha_start_election_async (Raft invariant) - ha.c: heartbeat_timeout_ms for hb_timeout_thread (not election_timeout_ms) - ha.c: clear voted_for before RDHUP jitter (stale vote from dead primary) - ha.c: same-term voted_for check in RDHUP jitter - config_repl.c/h: use xce_decode for upgradeConfig (capnp extraction before capn_free) - cluster.c: fix upgradeConfig handler direction (standbys apply, primaries skip) - cluster.c: remove raw_pkt_copy hack; use req directly - join.c: push config to newly-joined standbys - typemap.h/request.h/request.c: raw_sha256/raw_bytes fields for decoded upgradeConfig - test_ha_failover.c: update heartbeat timeout test to use g_fast_cfg
396 lines
12 KiB
C
396 lines
12 KiB
C
#if !defined(_TYPEMAP_H_)
|
|
|
|
#define _TYPEMAP_H_ 1
|
|
#include <stdint.h>
|
|
|
|
#include "clusterconfig.h"
|
|
#include "xce/types.h"
|
|
|
|
/* Forward declarations for config-replication types used in
|
|
* xce_reply_t and xce_request_t unions below. */
|
|
typedef struct repl_config_reply_s repl_config_reply_t;
|
|
typedef struct repl_config_push_reply_s repl_config_push_reply_t;
|
|
typedef struct upgrade_config_reply_s upgrade_config_reply_t;
|
|
typedef struct repl_config_req_s repl_config_req_t;
|
|
typedef struct repl_config_push_req_s repl_config_push_req_t;
|
|
typedef struct upgrade_config_req_s upgrade_config_req_t;
|
|
typedef struct heartbeat_msg_s heartbeat_msg_t;
|
|
typedef struct ctrl_hello_req_s ctrl_hello_req_t;
|
|
typedef struct ctrl_hello_reply_s ctrl_hello_reply_t;
|
|
|
|
typedef struct {
|
|
char* resname;
|
|
char* hostname;
|
|
} xce_reskey_t;
|
|
|
|
typedef struct {
|
|
int type;
|
|
union {
|
|
uint64_t jobid;
|
|
char* wname;
|
|
xce_reskey_t* reskey;
|
|
} condition;
|
|
} xce_match_t;
|
|
|
|
typedef struct {
|
|
int table;
|
|
int nrows;
|
|
void* rows;
|
|
} dataset_t;
|
|
|
|
typedef struct {
|
|
int role;
|
|
int data_which;
|
|
union {
|
|
hostattrs_t* attrs;
|
|
} data;
|
|
} xce_joindata_t;
|
|
|
|
typedef struct {
|
|
char* clustername;
|
|
int n_confvars;
|
|
xce_config_t** confvars;
|
|
int n_resources;
|
|
xce_resource_t** resources;
|
|
int n_queues;
|
|
xce_queue_t** queues;
|
|
int n_ugroups;
|
|
xce_group_config_t** ugroups;
|
|
int n_hgroups;
|
|
xce_group_config_t** hgroups;
|
|
int nresvals;
|
|
xce_name_val_t** resvals;
|
|
int nlimits;
|
|
xce_limit_config_t** limits;
|
|
int nhproviders;
|
|
scaler_config_t** hproviders;
|
|
} xce_conf4sched_t;
|
|
|
|
typedef struct {
|
|
char* clustername;
|
|
int n_confvars;
|
|
xce_config_t** confvars;
|
|
} xce_conf4worker_t;
|
|
|
|
typedef struct {
|
|
int data_which;
|
|
union {
|
|
xce_conf4sched_t* conf4sched;
|
|
xce_conf4worker_t* conf4worker;
|
|
cluster_config_t* clusterconfig;
|
|
} data;
|
|
} xce_configdata_t;
|
|
|
|
typedef struct {
|
|
int n_total;
|
|
int pos;
|
|
int count;
|
|
xce_job_t** jobs;
|
|
} xce_jobs_data_t;
|
|
|
|
typedef struct {
|
|
int n_total;
|
|
int pos;
|
|
int count;
|
|
xce_worker_t** workers;
|
|
} xce_workers_data_t;
|
|
|
|
typedef struct {
|
|
int n_total;
|
|
int pos;
|
|
int count;
|
|
xce_resource_t** resources;
|
|
} xce_resource_data_t;
|
|
|
|
typedef struct {
|
|
int n_total;
|
|
int pos;
|
|
int count;
|
|
xce_resval_t** vals;
|
|
} xce_resval_data_t;
|
|
|
|
typedef struct {
|
|
int count;
|
|
xce_resval_t** vals;
|
|
} xce_resval_list_t;
|
|
|
|
typedef struct {
|
|
uint64_t jobid;
|
|
int npids;
|
|
int* pids;
|
|
int npgids;
|
|
int* pgids;
|
|
int count;
|
|
xce_rusage_t** usages;
|
|
} xce_job_rusages_t;
|
|
|
|
typedef struct {
|
|
int table;
|
|
union {
|
|
xce_jobs_data_t* jobs;
|
|
xce_workers_data_t* workers;
|
|
xce_resource_data_t* resources;
|
|
xce_resval_data_t* resvals;
|
|
} data;
|
|
} xce_data_t;
|
|
|
|
typedef struct {
|
|
int table;
|
|
xce_match_t* condition;
|
|
} xce_get_t;
|
|
|
|
typedef struct {
|
|
int event;
|
|
union {
|
|
job_pidinfo_t* pidinfo;
|
|
job_failinfo_t* failinfo;
|
|
job_failinfo_t* dispatchfail;
|
|
xce_jobfinish_t* finishinfo;
|
|
uint64_t sched_start_job;
|
|
uint64_t sched_exit_job;
|
|
uint64_t sched_fail_job;
|
|
xce_job_t* sched_update_job;
|
|
xce_resval_list_t* sched_update_resvals;
|
|
uint64_t stopped_job;
|
|
uint64_t resumed_job;
|
|
xce_ckptinfo_t* ckptinfo;
|
|
xce_queue_t* sched_update_queue;
|
|
xce_task_pidinfo_t* taskinfo;
|
|
xce_task_failinfo_t* taskfailinfo;
|
|
xce_task_finish_t* taskfinish;
|
|
uint64_t pdone_job;
|
|
xce_job_rusages_t* sched_job_rusages;
|
|
xce_job_t* update_job;
|
|
} data;
|
|
} xce_notify_data_t;
|
|
|
|
typedef struct {
|
|
int table;
|
|
union {
|
|
xce_resval_list_t* resvals;
|
|
xce_job_rusages_t* jobrusages;
|
|
} data;
|
|
} xce_update_data_t;
|
|
|
|
typedef struct {
|
|
uint64_t term;
|
|
char *candidate_id;
|
|
uint64_t config_tag;
|
|
uint64_t last_log_lsn;
|
|
} request_vote_req_t;
|
|
|
|
typedef struct {
|
|
uint64_t term;
|
|
int vote_granted;
|
|
uint64_t peer_config_tag;
|
|
uint64_t peer_last_log_lsn;
|
|
} request_vote_reply_t;
|
|
|
|
typedef struct {
|
|
uint64_t seq;
|
|
uint32_t opcode;
|
|
int32_t status;
|
|
int data_which;
|
|
union {
|
|
char* debugmsg;
|
|
uint64_t jobid;
|
|
xce_data_t* getdata;
|
|
xce_clusterconf_t* clusterconf;
|
|
repl_config_reply_t* replconfigreplymsg;
|
|
repl_config_push_reply_t* replconfigpushreplymsg;
|
|
upgrade_config_reply_t* upgradeconfigreplymsg;
|
|
ctrl_hello_reply_t* ctrlhelloreplymsg;
|
|
request_vote_reply_t* requestvotereplymsg;
|
|
} data;
|
|
/* Req 7: standby redirect fields — populated when status == XCE_ERR_STANDBY */
|
|
char *redirect_host;
|
|
uint16_t redirect_port;
|
|
} xce_reply_t;
|
|
|
|
typedef struct {
|
|
uint32_t opcode;
|
|
uint64_t seq;
|
|
xce_peerinfo_t* from;
|
|
int body_which;
|
|
union {
|
|
xce_reply_t* replymsg;
|
|
char* hellomsg;
|
|
xce_joindata_t* joinmsg;
|
|
xce_configdata_t* configpushmsg;
|
|
xce_data_t* pushmsg;
|
|
xce_get_t* getmsg;
|
|
dataset_t* datamsg;
|
|
xce_job_dispatch_t* dispatchmsg;
|
|
xce_job_exec_t* jobstartmsg;
|
|
xce_notify_data_t* notifymsg;
|
|
xce_job_submit_t* jobsubmsg;
|
|
xce_update_data_t* updatemsg;
|
|
xce_jobctrl_t* jobctrlmsg;
|
|
xce_queuectrl_t* queuectrlmsg;
|
|
xce_taskrun_t* runtaskmsg;
|
|
xce_preason_set_t* preasonmsg;
|
|
xce_clear_reason_t* clrpreasonmsg;
|
|
xce_hostctrl_t* hostctrlmsg;
|
|
repl_config_req_t* replconfigrequestmsg;
|
|
repl_config_push_req_t* replconfigpushrequestmsg;
|
|
upgrade_config_req_t* upgradeconfigrequestmsg;
|
|
heartbeat_msg_t* heartbeatmsg;
|
|
ctrl_hello_req_t* ctrlhellomsg;
|
|
request_vote_req_t* requestvotemsg;
|
|
void* walcatchupentrymsg; /* wal_catchup_entry_t* */
|
|
void* snapshottransfermsg; /* snapshot_transfer_t* */
|
|
} body;
|
|
} xce_request_t;
|
|
|
|
/* ── HA election and replication Cap'n Proto mapped types ────────────────
|
|
* C-side counterparts for the HA protocol capnp structs added in the
|
|
* ha-config-bootstrap spec (controller.capnp structs with $C.mapname).
|
|
* Must be declared here so controller.capnp.h can reference them.
|
|
* request_vote_req_t and request_vote_reply_t are defined above
|
|
* xce_reply_t so they can be used in the xce_request_t body union. */
|
|
|
|
typedef struct {
|
|
char *name;
|
|
} probe_config_tag_req_t;
|
|
|
|
typedef struct {
|
|
uint64_t config_tag;
|
|
} probe_config_tag_reply_t;
|
|
|
|
typedef struct {
|
|
uint64_t term;
|
|
uint64_t lsn;
|
|
} wal_ack_req_t;
|
|
|
|
typedef struct {
|
|
uint64_t term;
|
|
uint64_t lsn;
|
|
int success;
|
|
} wal_ack_reply_t;
|
|
|
|
typedef struct {
|
|
uint64_t term;
|
|
uint64_t acked_tag;
|
|
int success;
|
|
} upgrade_config_ack_t;
|
|
|
|
typedef struct {
|
|
uint64_t snapshot_lsn;
|
|
xce_data_t *payload;
|
|
} snapshot_transfer_t;
|
|
|
|
typedef struct {
|
|
uint64_t lsn;
|
|
uint8_t entry_type;
|
|
xce_data_t *payload;
|
|
} wal_catchup_entry_t;
|
|
|
|
/* ── Persistence Cap'n Proto mapped types (P0-9) ─────────────────────
|
|
* These are the C-side counterparts for the capnp encode/decode shims.
|
|
* Named with capnp_ prefix to avoid collision with the plain-C structs
|
|
* in persist.h (persist_config_t) and the on-disk binary formats. */
|
|
typedef struct {
|
|
uint32_t walSyncMs;
|
|
uint32_t snapIntervalS;
|
|
uint32_t snapWalSizeMb;
|
|
uint32_t finishedRetainS;
|
|
uint32_t workerRetainS;
|
|
uint32_t syncTimeoutMs;
|
|
uint32_t snapKeepN;
|
|
uint32_t transientJobTimeoutS;
|
|
} capnp_persist_cfg_t;
|
|
|
|
typedef struct {
|
|
uint64_t lsn;
|
|
uint64_t term;
|
|
uint64_t timestamp;
|
|
uint32_t version;
|
|
uint32_t nJobs;
|
|
uint32_t nQueues;
|
|
uint32_t nWorkers;
|
|
uint32_t checksum;
|
|
uint32_t headerChecksum;
|
|
} capnp_snap_hdr_t;
|
|
|
|
typedef struct {
|
|
uint64_t term;
|
|
uint64_t lsn;
|
|
uint32_t entryType;
|
|
xce_data_t* payload; /* owned; decoded via decode_Data_ptr */
|
|
uint32_t crc;
|
|
uint32_t prevCrc;
|
|
} capnp_repl_envelope_t;
|
|
|
|
/* ── Config replication Cap'n Proto mapped types (Phase 1) ──────────────
|
|
* C-side counterparts for the capnp encode/decode shims for config sync.
|
|
* Struct tags match the forward declarations near the top of this file. */
|
|
|
|
struct repl_config_req_s {
|
|
uint64_t version_tag; /* local config version tag */
|
|
};
|
|
|
|
struct repl_config_reply_s {
|
|
int up_to_date; /* 1 if standby is current */
|
|
uint64_t version_tag; /* primary's current version tag */
|
|
xce_data_t *sha256; /* 32-byte SHA-256 of config blob (may be NULL) */
|
|
xce_data_t *bytes; /* serialised ClusterConfig blob (may be NULL) */
|
|
};
|
|
|
|
struct repl_config_push_req_s {
|
|
uint64_t version_tag; /* version tag of the pushed config */
|
|
xce_data_t *sha256; /* 32-byte SHA-256 */
|
|
xce_data_t *bytes; /* serialised ClusterConfig blob */
|
|
};
|
|
|
|
struct repl_config_push_reply_s {
|
|
int accepted; /* 1 if standby accepted the push */
|
|
uint64_t version_tag; /* standby's current version tag after decision */
|
|
};
|
|
|
|
struct upgrade_config_req_s {
|
|
/* Legacy decoded fields — populated by generated decode_UpgradeConfigRequest()
|
|
* via decode_Data_ptr. These are structurally broken for List(UInt8) wire
|
|
* fields (decode_Data_ptr reads a discriminant and misinterprets the bytes),
|
|
* but must remain for the generated code to compile. Do NOT use these for
|
|
* actual byte extraction. */
|
|
xce_data_t *sha256; /* broken: always a bogus xce_jobs_data_t */
|
|
xce_data_t *bytes; /* broken: always a bogus xce_jobs_data_t */
|
|
|
|
/* Correct raw bytes extracted by xce_decode() before capn_free() runs,
|
|
* using UpgradeConfigRequest_get_sha256/bytes + capn_getv8.
|
|
* Both are malloc'd and owned here; freed by free_upgrade_config_req_raw()
|
|
* called from free_Request() after free_UpgradeConfigRequest().
|
|
* raw_sha256 is exactly 32 bytes when non-NULL. */
|
|
uint8_t *raw_sha256; /* 32-byte SHA-256, malloc'd, or NULL */
|
|
uint8_t *raw_bytes; /* serialised ClusterConfig blob, malloc'd, or NULL */
|
|
size_t raw_bytes_len; /* length of raw_bytes */
|
|
};
|
|
|
|
struct upgrade_config_reply_s {
|
|
uint64_t version_tag; /* new version tag after upgrade */
|
|
char *error; /* NULL on success; error message on failure */
|
|
uint32_t acked_count; /* number of standbys that acknowledged */
|
|
};
|
|
|
|
struct heartbeat_msg_s {
|
|
uint32_t sender_priority; /* sender's HA priority */
|
|
uint64_t term; /* sender's current term */
|
|
uint64_t committed_lsn; /* Req 5: primary's committed WAL LSN */
|
|
uint64_t committed_config_tag; /* Req 8: primary's committed config tag */
|
|
};
|
|
|
|
/* CtrlHello state sync (Req 6.5, 6.6, 6.7) */
|
|
struct ctrl_hello_req_s {
|
|
char *name; /* standby's logical name */
|
|
xce_data_t *key; /* SHA-256(cluster_name + shared_secret) auth key */
|
|
uint64_t applied_lsn; /* standby's last applied WAL LSN */
|
|
uint64_t config_tag; /* standby's current config tag */
|
|
};
|
|
|
|
struct ctrl_hello_reply_s {
|
|
uint64_t committed_lsn; /* primary's committed WAL LSN */
|
|
int needs_snapshot; /* 1 if standby LSN < oldest WAL segment */
|
|
uint64_t config_tag; /* primary's current config tag */
|
|
};
|
|
|
|
#endif
|