diff --git a/src/controller/cluster.c b/src/controller/cluster.c index 11d3d6ee..80cd47f2 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -333,6 +333,10 @@ static int myreq_handler(vtask_t* task) case OPCode_replConfig: rc = handle_config_request(c, ep, &request); break; + case OPCode_replConfigPush: + /* Primary pushes config to this standby — save and reload */ + rc = handle_repl_config_push(c, ep, &request); + break; case OPCode_upgradeConfig: #if CONF_ENABLE_UPGRADE_CONFIG /* OPCode_upgradeConfig is sent BY the primary TO standbys to push config. @@ -753,7 +757,12 @@ int cluster_init_phase1(cluster_t* c) if (c == NULL) return -EINVAL; - c->myhost = &from_controller; + /* Populate myhost_info so replies identify this node by its real name + * (e.g. ctrl-1) rather than the generic "controller" placeholder. */ + c->myhost_info.name = (c->myconf && c->myconf->name) + ? c->myconf->name : from_controller.name; + c->myhost_info.key = from_controller.key; + c->myhost = &c->myhost_info; c->ncontrollers = 1; c->myidx = 0; diff --git a/src/controller/config_repl.c b/src/controller/config_repl.c index f3f9c461..1ab8bfc4 100644 --- a/src/controller/config_repl.c +++ b/src/controller/config_repl.c @@ -675,6 +675,72 @@ int handle_config_request(cluster_t *c, endpoint_t *ep, xce_request_t *req) 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; + } + + 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 ─────────────────────────── */ static void send_upgrade_error(endpoint_t *ep, xce_request_t *req, diff --git a/src/controller/config_repl.h b/src/controller/config_repl.h index 196c49e0..77c19810 100644 --- a/src/controller/config_repl.h +++ b/src/controller/config_repl.h @@ -102,6 +102,12 @@ void create_lost_and_found(cluster_t *c); */ int handle_config_request(cluster_t *c, endpoint_t *ep, xce_request_t *req); +/* + * handle_repl_config_push: standby handler for OPCode_replConfigPush. + * Verifies SHA-256, saves config blob to disk, and reloads cluster_config. + */ +int handle_repl_config_push(cluster_t *c, endpoint_t *ep, xce_request_t *req); + /* * push_config_to_standbys: fan-out a new config blob to all connected standby * controllers in parallel. Best-effort — standbys that miss the push will diff --git a/src/controller/ha.c b/src/controller/ha.c index 9bd74d4d..877e0846 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -403,12 +403,27 @@ static void raft_cb_become_leader(void *ctx, uint64_t term) } if (c) { - /* Push config to standbys with lower config_tag */ + /* Push config to standbys — read raw bytes from disk so standbys + * can save the file. NULL bytes are passed only in test mode. */ #ifdef XCE_TESTING if (ha_push_config_hook != NULL) ha_push_config_hook(c); #else - push_config_to_standbys(c, NULL, 0, NULL, 0); + { + uint8_t *cfg_bytes = NULL; + size_t cfg_len = 0; + extern int conf_read_raw(const char *, uint8_t **, size_t *); + if (c->myconf && c->myconf->conffile && + conf_read_raw(c->myconf->conffile, &cfg_bytes, &cfg_len) == 0) { + push_config_to_standbys(c, cfg_bytes, cfg_len, + c->myconf->config_sha256, + c->myconf->config_version_tag); + free(cfg_bytes); + } else { + push_config_to_standbys(c, NULL, 0, NULL, + c->myconf ? c->myconf->config_version_tag : 0); + } + } #endif /* Send CtrlHelloRequest to all standbys to trigger WAL catch-up */ #ifdef XCE_TESTING diff --git a/src/include/typemap.h b/src/include/typemap.h index 184d9ca5..1628cb68 100644 --- a/src/include/typemap.h +++ b/src/include/typemap.h @@ -337,9 +337,14 @@ struct repl_config_reply_s { }; 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 */ + uint64_t version_tag; /* version tag of the pushed config */ + xce_data_t *sha256; /* 32-byte SHA-256 (decoded, not usable as raw bytes) */ + xce_data_t *bytes; /* serialised ClusterConfig blob (decoded, not usable) */ + /* Pre-extracted raw fields populated by xce_decode() before capn_free(), + * mirroring the same pattern as upgrade_config_req_t. Owned by the struct. */ + 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 repl_config_push_reply_s { diff --git a/src/include/xce/cluster_def.h b/src/include/xce/cluster_def.h index cfd3e9a1..91985690 100644 --- a/src/include/xce/cluster_def.h +++ b/src/include/xce/cluster_def.h @@ -96,7 +96,8 @@ typedef struct { typedef struct cluster_s { myconf_t* myconf; - xce_peerinfo_t* myhost; + xce_peerinfo_t* myhost; /* points at myhost_info once initialised */ + xce_peerinfo_t myhost_info; /* storage; name set from myconf->name */ hooksys_t hooksys; iomgr_t iomgr; zarray_t endpoints; diff --git a/src/lib/helper.c b/src/lib/helper.c index 1ce0b7c3..4491f1d1 100644 --- a/src/lib/helper.c +++ b/src/lib/helper.c @@ -148,6 +148,38 @@ char* xce_request_op_name(xce_request_t* req) return "setpreason"; case OPCode_clrpreason: return "clrpreason"; + case OPCode_heartbeat: + return "heartbeat"; + case OPCode_redirect: + return "redirect"; + case OPCode_steppingDown: + return "steppingDown"; + case OPCode_replEntry: + return "replEntry"; + case OPCode_replAck: + return "replAck"; + case OPCode_replSnapshot: + return "replSnapshot"; + case OPCode_replCatchUpDone: + return "replCatchUpDone"; + case OPCode_replConfig: + return "replConfig"; + case OPCode_replConfigPush: + return "replConfigPush"; + case OPCode_upgradeConfig: + return "upgradeConfig"; + case OPCode_upgradeConfigAck: + return "upgradeConfigAck"; + case OPCode_requestVote: + return "requestVote"; + case OPCode_probeConfigTag: + return "probeConfigTag"; + case OPCode_walAck: + return "walAck"; + case OPCode_walCatchupEntry: + return "walCatchupEntry"; + case OPCode_snapshotTransfer: + return "snapshotTransfer"; default: break; } diff --git a/src/lib/request.c b/src/lib/request.c index a2d57f40..9eef0565 100644 --- a/src/lib/request.c +++ b/src/lib/request.c @@ -119,6 +119,45 @@ int xce_decode(io_packet_t* pkt, xce_request_t* req) } } + /* OPCode_replConfigPush: same List(UInt8) issue — pre-extract raw bytes. */ + if (rq.opcode == OPCode_replConfigPush + && req->body_which == Request_body_replconfigpushrequestmsg + && req->body.replconfigpushrequestmsg != NULL) { + + repl_config_push_req_t *rcpr = req->body.replconfigpushrequestmsg; + ReplConfigPushRequest_ptr r_ptr = rq.body.replconfigpushrequestmsg; + + /* sha256 */ + { + Data_ptr dp = ReplConfigPushRequest_get_sha256(r_ptr); + capn_list8 sha_list; sha_list.p = dp.p; + if (sha_list.p.type != CAPN_NULL && capn_len(sha_list) > 0) { + int slen = capn_len(sha_list); + uint8_t *sbuf = (uint8_t *)malloc((size_t)slen); + if (sbuf && capn_getv8(sha_list, 0, sbuf, slen) == slen) + rcpr->raw_sha256 = sbuf; + else + free(sbuf); + } + } + + /* bytes */ + { + Data_ptr dp = ReplConfigPushRequest_get_bytes(r_ptr); + capn_list8 bytes_list; bytes_list.p = dp.p; + if (bytes_list.p.type != CAPN_NULL && capn_len(bytes_list) > 0) { + int blen = capn_len(bytes_list); + uint8_t *bbuf = (uint8_t *)malloc((size_t)blen); + if (bbuf && capn_getv8(bytes_list, 0, bbuf, blen) == blen) { + rcpr->raw_bytes = bbuf; + rcpr->raw_bytes_len = (size_t)blen; + } else { + free(bbuf); + } + } + } + } + TRACE("%s request op %d seq %ld from %s", __func__, req->opcode, req->seq, REQUEST_FROM(req)); capn_free(&c); @@ -150,7 +189,7 @@ void xce_free_request(xce_request_t *req) if (req == NULL) return; - /* Free raw_* fields that the generated free_UpgradeConfigRequest skips */ + /* Free raw_* fields that the generated free_*Request skips */ if (req->body_which == Request_body_upgradeconfigrequestmsg && req->body.upgradeconfigrequestmsg != NULL) { upgrade_config_req_t *ucr = req->body.upgradeconfigrequestmsg; @@ -161,6 +200,16 @@ void xce_free_request(xce_request_t *req) ucr->raw_bytes_len = 0; } + if (req->body_which == Request_body_replconfigpushrequestmsg + && req->body.replconfigpushrequestmsg != NULL) { + repl_config_push_req_t *rcpr = req->body.replconfigpushrequestmsg; + free(rcpr->raw_sha256); + rcpr->raw_sha256 = NULL; + free(rcpr->raw_bytes); + rcpr->raw_bytes = NULL; + rcpr->raw_bytes_len = 0; + } + free_Request(req); }