feat(config-repl): add raw capnp shims for OP_replConfig wire protocol

xce_replconfig_reply_raw() - primary sends reply with real bytes via endpoint
xce_replconfig_req_ssl() - standby sends request over raw SSL socket
xce_replconfig_reply_ssl_decode() - standby decodes reply from raw SSL
xce_upgrade_config_req_extract() - documents architectural limitation; returns -ENOTSUP
xce_upgrade_config_req_extract_raw() - extract bytes from raw packet for OP_upgradeConfig

Same bypass pattern as existing xce_replconfigpush_req_raw(): build the
capnp message directly via capn_new_list8 + capn_setv8, sidestepping the
encode_Data_ptr path that routes raw bytes through xce_data_t (a
discriminated cluster-data union that cannot carry raw byte arrays).

Unblocks handle_config_request() and do_one_config_fetch() wiring.

Task A research finding: xce_upgrade_config_req_extract(xce_request_t*)
is unimplementable because xce_decode() calls capn_free() before any
handler runs, and decode_Data_ptr converts the wire-format List(UInt8)
into a zero-filled xce_data_t. Task D must capture the raw io_packet_t
bytes before xce_decode tears them down, then call the _raw variant.
This commit is contained in:
Rongsong Shen 2026-05-21 13:02:49 +00:00
parent f3babec5bc
commit 26393e919c
3 changed files with 848 additions and 0 deletions

View file

@ -2,9 +2,12 @@
#include <stdarg.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
#include <openssl/ssl.h>
#include "mylock.h"
#include "netio.h"
#include "request.h"
@ -224,6 +227,483 @@ int xce_replconfigpush_req_raw(endpoint_t *ep, xce_peerinfo_t *from,
return 0;
}
/*
* xce_replconfig_reply_raw: primary sends OP_replConfig reply with real
* config bytes via an endpoint (uses iomgr queueing).
*
* Bypasses encode_ReplConfigReply_ptr (which routes sha256/bytes through
* xce_data_t — a discriminated cluster-data union that cannot carry raw
* bytes). Constructs Request { opcode=OPCode_reply, body=replymsg }
* with RequestReply { data=replconfigreplymsg } directly via the capnp
* wire-format API (capn_new_list8 + capn_setv8), then enqueues for send.
*/
int xce_replconfig_reply_raw(endpoint_t *ep, xce_peerinfo_t *from,
uint64_t seq, uint64_t tag,
const uint8_t *sha256, size_t sha256_len,
const uint8_t *config, size_t config_len)
{
struct capn c;
struct capn_segment *cs;
struct Request rq;
struct RequestReply rr;
struct ReplConfigReply reply_struct;
ReplConfigReply_ptr reply_ptr;
RequestReply_ptr rr_ptr;
Request_ptr rp;
io_packet_t *pkt;
int rc;
if (ep == NULL)
return -EINVAL;
capn_init_malloc(&c);
cs = capn_root(&c).seg;
memset(&rq, 0, sizeof(rq));
memset(&rr, 0, sizeof(rr));
memset(&reply_struct, 0, sizeof(reply_struct));
/* ReplConfigReply fields */
reply_struct.upToDate = 0;
reply_struct.versionTag = tag;
if (sha256 && sha256_len > 0) {
capn_list8 sha_list = capn_new_list8(cs, (int)sha256_len);
capn_setv8(sha_list, 0, sha256, (int)sha256_len);
reply_struct.sha256.p = sha_list.p;
}
if (config && config_len > 0) {
capn_list8 bytes_list = capn_new_list8(cs, (int)config_len);
capn_setv8(bytes_list, 0, config, (int)config_len);
reply_struct.bytes.p = bytes_list.p;
}
reply_ptr = new_ReplConfigReply(cs);
write_ReplConfigReply(&reply_struct, reply_ptr);
/* RequestReply envelope */
rr.seq = seq;
rr.opcode = OPCode_replConfig; /* opcode being replied to */
rr.status = 0;
rr.data_which = RequestReply_data_replconfigreplymsg;
rr.data.replconfigreplymsg = reply_ptr;
rr_ptr = new_RequestReply(cs);
write_RequestReply(&rr, rr_ptr);
/* Outer Request */
rq.opcode = OPCode_reply;
rq.seq = seq;
rq.body_which = Request_body_replymsg;
rq.body.replymsg = rr_ptr;
encode_PeerInfo_ptr(cs, &rq.from, from);
rp = new_Request(cs);
write_Request(&rq, rp);
capn_setp(capn_root(&c), 0, rp.p);
pkt = (io_packet_t *)calloc(1, sizeof(*pkt));
if (!pkt) {
capn_free(&c);
return -ENOMEM;
}
pkt->sz = capn_size(&c);
rc = alloc_packet(pkt);
if (rc < 0) {
free(pkt);
capn_free(&c);
return -errno;
}
capn_write_mem(&c, (uint8_t *)pkt->data, pkt->sz, 0);
capn_free(&c);
endpoint_enq(ep, &(pkt->link));
return 0;
}
/*
* xce_replconfig_req_ssl: standby sends OP_replConfig over a raw SSL
* socket (no iomgr). Used during early-boot config fetch from primary
* before the iomgr is fully wired.
*
* Wire framing matches netio.c: uint64_t size prefix (host byte order)
* followed by `size` bytes of capnp data.
*/
int xce_replconfig_req_ssl(SSL *ssl, uint64_t version_tag)
{
struct capn c;
struct capn_segment *cs;
struct Request rq;
struct ReplConfigRequest req_struct;
ReplConfigRequest_ptr req_ptr;
Request_ptr rp;
uint8_t *buf = NULL;
int64_t blen = 0;
int rc = 0;
if (ssl == NULL)
return -EINVAL;
capn_init_malloc(&c);
cs = capn_root(&c).seg;
memset(&rq, 0, sizeof(rq));
memset(&req_struct, 0, sizeof(req_struct));
req_struct.versionTag = version_tag;
req_ptr = new_ReplConfigRequest(cs);
write_ReplConfigRequest(&req_struct, req_ptr);
rq.opcode = OPCode_replConfig;
rq.seq = 0;
rq.body_which = Request_body_replconfigrequestmsg;
rq.body.replconfigrequestmsg = req_ptr;
/* from = NULL (no peerinfo for one-shot SSL fetch) */
encode_PeerInfo_ptr(cs, &rq.from, NULL);
rp = new_Request(cs);
write_Request(&rq, rp);
capn_setp(capn_root(&c), 0, rp.p);
blen = capn_size(&c);
if (blen <= 0) {
capn_free(&c);
return -EIO;
}
buf = (uint8_t *)malloc((size_t)blen);
if (!buf) {
capn_free(&c);
return -ENOMEM;
}
if (capn_write_mem(&c, buf, (size_t)blen, 0) < 0) {
free(buf);
capn_free(&c);
return -EIO;
}
capn_free(&c);
/* Wire framing: 8-byte size header then payload (matches netio.c
* io_fsm_perform: writes &pkt->sz raw, then pkt->data). */
{
uint64_t hdr = (uint64_t)blen;
int nsent = 0;
nsent = SSL_write(ssl, &hdr, (int)sizeof(hdr));
if (nsent != (int)sizeof(hdr)) {
rc = -EIO;
goto out;
}
nsent = SSL_write(ssl, buf, (int)blen);
if (nsent != (int)blen) {
rc = -EIO;
goto out;
}
}
out:
free(buf);
return rc;
}
/* Helper: read exactly `len` bytes from SSL, looping on partial reads. */
static int ssl_read_all(SSL *ssl, void *dst, size_t len)
{
size_t off = 0;
while (off < len) {
int n = SSL_read(ssl, (char *)dst + off, (int)(len - off));
if (n <= 0) {
return -EIO;
}
off += (size_t)n;
}
return 0;
}
/*
* xce_replconfig_reply_ssl_decode: standby reads the primary's
* OP_replConfig reply from a raw SSL socket and extracts up_to_date,
* version_tag, sha256 and bytes.
*
* Bypasses xce_decode (which would route sha256/bytes through
* decode_Data_ptr → xce_data_t and lose the raw byte content).
* Instead, reads the capnp message directly into capn, walks down
* Request → RequestReply → ReplConfigReply, and pulls sha256/bytes
* via capn_len + capn_getv8 on the raw capn_list8.
*
* Output buffers (*out_sha256 and *out_bytes) are malloc'd and must
* be freed by the caller. Both may be NULL on a fast path
* (up_to_date == 1). Caller is responsible for distinguishing.
*/
int xce_replconfig_reply_ssl_decode(SSL *ssl,
int *out_up_to_date,
uint64_t *out_tag,
uint8_t **out_sha256,
uint8_t **out_bytes,
size_t *out_len)
{
struct capn c;
Request_ptr rp;
struct Request rq;
RequestReply_ptr rr_ptr;
struct RequestReply rr;
ReplConfigReply_ptr reply_ptr;
struct ReplConfigReply reply_struct;
capn_list8 sha_list;
capn_list8 bytes_list;
uint64_t hdr = 0;
uint8_t *buf = NULL;
int rc = 0;
if (ssl == NULL || out_up_to_date == NULL || out_tag == NULL
|| out_sha256 == NULL || out_bytes == NULL || out_len == NULL) {
return -EINVAL;
}
*out_up_to_date = 0;
*out_tag = 0;
*out_sha256 = NULL;
*out_bytes = NULL;
*out_len = 0;
rc = ssl_read_all(ssl, &hdr, sizeof(hdr));
if (rc < 0) {
return rc;
}
if (hdr == 0 || hdr > (uint64_t)(64 * 1024 * 1024)) {
/* 64 MiB sanity ceiling */
return -EMSGSIZE;
}
buf = (uint8_t *)malloc((size_t)hdr);
if (!buf) {
return -ENOMEM;
}
rc = ssl_read_all(ssl, buf, (size_t)hdr);
if (rc < 0) {
free(buf);
return rc;
}
if (capn_init_mem(&c, buf, (size_t)hdr, 0) < 0) {
free(buf);
return -EPROTO;
}
rp.p = capn_getp(capn_root(&c), 0, 1);
read_Request(&rq, rp);
if (rq.body_which != Request_body_replymsg) {
rc = -EPROTO;
goto cleanup;
}
rr_ptr = rq.body.replymsg;
read_RequestReply(&rr, rr_ptr);
if (rr.data_which != RequestReply_data_replconfigreplymsg) {
rc = -EPROTO;
goto cleanup;
}
reply_ptr = rr.data.replconfigreplymsg;
read_ReplConfigReply(&reply_struct, reply_ptr);
*out_up_to_date = (int)reply_struct.upToDate;
*out_tag = reply_struct.versionTag;
/* Pull sha256 and bytes via raw capn_list8 access (NOT decode_Data
* which would corrupt via xce_data_t). Data_ptr and capn_list8
* share {capn_ptr p} layout. */
sha_list.p = reply_struct.sha256.p;
bytes_list.p = reply_struct.bytes.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) {
rc = -ENOMEM;
goto cleanup;
}
if (capn_getv8(sha_list, 0, sbuf, slen) != slen) {
free(sbuf);
rc = -EPROTO;
goto cleanup;
}
*out_sha256 = sbuf;
}
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) {
free(*out_sha256);
*out_sha256 = NULL;
rc = -ENOMEM;
goto cleanup;
}
if (capn_getv8(bytes_list, 0, bbuf, blen) != blen) {
free(bbuf);
free(*out_sha256);
*out_sha256 = NULL;
rc = -EPROTO;
goto cleanup;
}
*out_bytes = bbuf;
*out_len = (size_t)blen;
}
cleanup:
capn_free(&c);
free(buf);
return rc;
}
/*
* xce_upgrade_config_req_extract: attempt to extract sha256 + bytes
* from a decoded OP_upgradeConfig request.
*
* ARCHITECTURAL FINDING (Task A research):
* --------------------------------------------------------------------
* The decoded form is unusable for byte extraction. When iomgr receives
* an OP_upgradeConfig packet, xce_decode() calls decode_Request →
* decode_UpgradeConfigRequest → decode_Data_ptr. decode_Data_ptr
* allocates a fresh xce_data_t and walks decode_Data, which treats the
* pointer as a discriminated Data union (reading a 16-bit discriminator
* at offset 0). But the wire actually carries a List(UInt8) with
* datasz=0, so capn_read16 returns 0 (Data_data_schedjobs) and a
* zero-filled xce_jobs_data_t is allocated. No raw bytes survive.
*
* Worse: xce_decode() then calls capn_free() and free_packet(), so the
* original capnp segment containing the List(UInt8) bytes is gone by
* the time any handler runs. The xce_data_t* fields on the decoded
* xce_request_t are non-NULL but contain only the bogus jobs_data_t.
*
* This function therefore always returns -ENOTSUP. Task D must use
* xce_upgrade_config_req_extract_raw() below and arrange for the raw
* io_packet_t bytes to reach handle_upgrade_config() before xce_decode
* destroys them — e.g. by adding a pre-decode peek for OP_upgradeConfig
* in the message-handler dispatch, or by stashing the raw buffer on
* xce_request_t for select opcodes.
*/
int xce_upgrade_config_req_extract(const xce_request_t *req,
uint8_t **out_bytes, size_t *out_len,
uint8_t **out_sha256)
{
if (req == NULL || out_bytes == NULL || out_len == NULL
|| out_sha256 == NULL) {
return -EINVAL;
}
*out_bytes = NULL;
*out_len = 0;
*out_sha256 = NULL;
return -ENOTSUP;
}
/*
* xce_upgrade_config_req_extract_raw: parse OP_upgradeConfig request
* bytes from a raw on-wire packet and extract sha256 + config bytes.
*
* Use this from handle_upgrade_config() once a path is wired to capture
* the io_packet_t->data buffer before xce_decode() tears it down (see
* the xce_upgrade_config_req_extract() comment for why the decoded
* form is unusable).
*
* Output buffers (*out_bytes, *out_sha256) are malloc'd and owned by
* the caller.
*/
int xce_upgrade_config_req_extract_raw(const uint8_t *pkt_data, size_t pkt_sz,
uint8_t **out_bytes, size_t *out_len,
uint8_t **out_sha256)
{
struct capn c;
Request_ptr rp;
struct Request rq;
UpgradeConfigRequest_ptr u_ptr;
struct UpgradeConfigRequest u_struct;
capn_list8 sha_list;
capn_list8 bytes_list;
int rc = 0;
if (pkt_data == NULL || pkt_sz == 0 || out_bytes == NULL
|| out_len == NULL || out_sha256 == NULL) {
return -EINVAL;
}
*out_bytes = NULL;
*out_len = 0;
*out_sha256 = NULL;
if (capn_init_mem(&c, pkt_data, pkt_sz, 0) < 0) {
return -EPROTO;
}
rp.p = capn_getp(capn_root(&c), 0, 1);
read_Request(&rq, rp);
if (rq.body_which != Request_body_upgradeconfigrequestmsg) {
rc = -EPROTO;
goto cleanup;
}
u_ptr = rq.body.upgradeconfigrequestmsg;
read_UpgradeConfigRequest(&u_struct, u_ptr);
sha_list.p = u_struct.sha256.p;
bytes_list.p = u_struct.bytes.p;
if (sha_list.p.type == CAPN_NULL || capn_len(sha_list) <= 0
|| bytes_list.p.type == CAPN_NULL || capn_len(bytes_list) <= 0) {
rc = -EPROTO;
goto cleanup;
}
{
int slen = capn_len(sha_list);
int blen = capn_len(bytes_list);
uint8_t *sbuf;
uint8_t *bbuf;
sbuf = (uint8_t *)malloc((size_t)slen);
if (!sbuf) {
rc = -ENOMEM;
goto cleanup;
}
if (capn_getv8(sha_list, 0, sbuf, slen) != slen) {
free(sbuf);
rc = -EPROTO;
goto cleanup;
}
bbuf = (uint8_t *)malloc((size_t)blen);
if (!bbuf) {
free(sbuf);
rc = -ENOMEM;
goto cleanup;
}
if (capn_getv8(bytes_list, 0, bbuf, blen) != blen) {
free(sbuf);
free(bbuf);
rc = -EPROTO;
goto cleanup;
}
*out_sha256 = sbuf;
*out_bytes = bbuf;
*out_len = (size_t)blen;
}
cleanup:
capn_free(&c);
return rc;
}
int xce_nop_req(endpoint_t* ep)
{
xce_request_t request