#include <errno.h>
#include <stdarg.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
#include "xce/errnox.h"  /* E_XCE_NOT_READY, E_XCE_STANDBY */

#include <openssl/ssl.h>

#include "mylock.h"
#include "netio.h"
#include "request.h"
#include "typemap.h"
#include "utils.h"
#include "xce/types.h"

#define REQUEST_FROM(req)                                                      \
    ((req)->from == NULL)                                                      \
        ? "Unknown"                                                            \
        : (((req)->from->name == NULL) ? "Unknown" : (req->from->name))

int xce_encode(xce_request_t* req, io_packet_t* pkt)
{
    struct capn          c;
    struct Request       rq;
    struct capn_segment* cs;
    Request_ptr          rp;
    int                  rc;

    TRACE("%s request op %d seq %ld from %s", __func__, req->opcode, req->seq,
          REQUEST_FROM(req));

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;

    encode_Request(cs, &rq, req);

    cs = capn_root(&c).seg;
    rp = new_Request(cs);
    write_Request(&rq, rp);

    capn_setp(capn_root(&c), 0, rp.p);

    pkt->sz = capn_size(&c);
    rc      = alloc_packet(pkt);
    if (rc < 0) {
        capn_free(&c);
        return -errno;
    }

    capn_write_mem(&c, (uint8_t*)pkt->data, pkt->sz, 0);
    capn_free(&c);

    return 1;
}

/* extract_data_bytes: copy a capnp Data (List(UInt8)) field into a fresh
 * malloc'd buffer.  Returns NULL (*out_len = 0) for absent/empty fields or
 * allocation failure.  Used by xce_decode() to pre-extract raw payload
 * bytes before capn_free() destroys the segment. */
static uint8_t *extract_data_bytes(capn_data dp, size_t *out_len)
{
    *out_len = 0;

    capn_list8 l;
    l.p = dp.p;
    if (l.p.type == CAPN_NULL)
        return NULL;

    int len = capn_len(l);
    if (len <= 0)
        return NULL;

    uint8_t *buf = (uint8_t *)malloc((size_t)len);
    if (!buf)
        return NULL;
    if (capn_getv8(l, 0, buf, len) != len) {
        free(buf);
        return NULL;
    }

    *out_len = (size_t)len;
    return buf;
}

int xce_decode(io_packet_t* pkt, xce_request_t* req)
{
    struct capn    c;
    struct Request rq;
    Request_ptr    rp;

    capn_init_mem(&c, (const uint8_t*)pkt->data, pkt->sz, 0);
    rp.p = capn_getp(capn_root(&c), 0, 1);

    read_Request(&rq, rp);

    decode_Request(req, &rq);

    /* OPCode_upgradeConfig carries sha256 and bytes as List(UInt8) on the wire.
     * decode_UpgradeConfigRequest() intentionally leaves raw_sha256/raw_bytes as
     * NULL (the project's Data union cannot carry raw bytes via decode_XceData_ptr).
     * Extract them here using UpgradeConfigRequest_get_* + capn_getv8 BEFORE
     * capn_free() destroys the capnp segment. */
    if (rq.opcode == OPCode_upgradeConfig
        && req->body_which == Request_body_upgradeconfigrequestmsg
        && req->body.upgradeconfigrequestmsg != NULL) {

        upgrade_config_req_t *ucr = req->body.upgradeconfigrequestmsg;
        UpgradeConfigRequest_ptr u_ptr = rq.body.upgradeconfigrequestmsg;

        /* sha256 */
        {
            capn_data    dp        = UpgradeConfigRequest_get_sha256(u_ptr);
            capn_list8  sha_list;
            sha_list.p = dp.p;
            if (sha_list.p.type != CAPN_NULL) {
                int slen = capn_len(sha_list);
                if (slen > 0) {
                    uint8_t *sbuf = (uint8_t *)malloc((size_t)slen);
                    if (sbuf && capn_getv8(sha_list, 0, sbuf, slen) == slen) {
                        ucr->raw_sha256 = sbuf;
                    } else {
                        free(sbuf);
                    }
                }
            }
        }

        /* bytes */
        {
            capn_data    dp         = UpgradeConfigRequest_get_bytes(u_ptr);
            capn_list8  bytes_list;
            bytes_list.p = dp.p;
            if (bytes_list.p.type != CAPN_NULL) {
                int blen = capn_len(bytes_list);
                if (blen > 0) {
                    uint8_t *bbuf = (uint8_t *)malloc((size_t)blen);
                    if (bbuf && capn_getv8(bytes_list, 0, bbuf, blen) == blen) {
                        ucr->raw_bytes     = bbuf;
                        ucr->raw_bytes_len = (size_t)blen;
                    } else {
                        free(bbuf);
                    }
                }
            }
        }
    }

    /* 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 */
        {
            capn_data   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 */
        {
            capn_data   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);
                }
            }
        }
    }

    /* OPCode_walCatchupEntry: payload :Data per (sub-)entry — pre-extract
     * raw bytes into raw_payload/raw_payload_len (replConfigPush pattern).
     * The wire entries list order matches the decoded entries[] order. */
    if (rq.opcode == OPCode_walCatchupEntry
        && req->body_which == Request_body_walcatchupentrymsg
        && req->body.walcatchupentrymsg != NULL) {

        wal_catchup_entry_t *batch = req->body.walcatchupentrymsg;
        WalCatchupEntry_ptr  b_ptr = rq.body.walcatchupentrymsg;

        /* Top-level payload (single-entry mode) */
        batch->raw_payload =
            extract_data_bytes(WalCatchupEntry_get_payload(b_ptr),
                               &batch->raw_payload_len);

        /* bug-847: the top-level batch_payloads blob — every sub-entry's
         * payload concatenated in ONE Data field (see wal_entry_to_wire /
         * xce_wal_batch_encode_raw). A single top-level Data blob does not
         * exhibit the per-element cross-segment pointer loss that scattered
         * per-sub-entry Data blobs did. */
        batch->raw_batch_payloads =
            extract_data_bytes(WalCatchupEntry_get_batchPayloads(b_ptr),
                               &batch->raw_batch_payloads_len);

        /* ha-term-safety task 3.2: the top-level container's AppendEntries
         * metadata (term, prevLogIndex, prevLogTerm, leaderCommit) is decoded
         * into batch->{term,prev_log_index,prev_log_term,leader_commit} by the
         * generated decode_WalCatchupEntry (all scalar $C.mapname fields) — only
         * the Data payload needs manual extraction (above/below). The standby
         * intake (ha_consume_append_entries) reads those struct fields directly. */

        /* Batch sub-entries: each carries its own lsn/entryType/term/
         * wire_payload_len (scalar, decoded — bug-847 immune) + payload. Slice
         * this entry's bytes out of the top-level batch_payloads blob using
         * wire_payload_len and a running cumulative offset (entries stream in
         * the same order they were concatenated on encode). Falls back to the
         * legacy per-element Data field only when no batch_payloads blob was
         * sent at all (defensive — every current sender populates it). */
        WalCatchupEntry_list wlist = WalCatchupEntry_get_entries(b_ptr);
        int wn = (wlist.p.type != CAPN_NULL) ? capn_len(wlist) : 0;
        size_t batch_off = 0;
        for (int i = 0; i < wn && i < batch->n_entries; i++) {
            if (!batch->entries || !batch->entries[i])
                continue;
            struct WalCatchupEntry we;
            get_WalCatchupEntry(&we, wlist, i);

            if (batch->raw_batch_payloads != NULL) {
                uint32_t plen = batch->entries[i]->wire_payload_len;
                if (plen > 0 &&
                    batch_off + plen <= batch->raw_batch_payloads_len) {
                    uint8_t *slice = (uint8_t *)malloc(plen);
                    if (slice) {
                        memcpy(slice, batch->raw_batch_payloads + batch_off,
                              plen);
                        batch->entries[i]->raw_payload     = slice;
                        batch->entries[i]->raw_payload_len = plen;
                    }
                    batch_off += plen;
                }
            } else {
                batch->entries[i]->raw_payload =
                    extract_data_bytes(we.payload,
                                       &batch->entries[i]->raw_payload_len);
            }
        }
    }

    /* OPCode_snapshotTransfer: chunk payload :Data — pre-extract raw bytes. */
    if (rq.opcode == OPCode_snapshotTransfer
        && req->body_which == Request_body_snapshottransfermsg
        && req->body.snapshottransfermsg != NULL) {

        snapshot_transfer_t *chunk = req->body.snapshottransfermsg;
        SnapshotTransfer_ptr s_ptr = rq.body.snapshottransfermsg;
        chunk->raw_payload =
            extract_data_bytes(SnapshotTransfer_get_payload(s_ptr),
                               &chunk->raw_payload_len);
    }

    TRACE("%s request op %d seq %ld from %s", __func__, req->opcode, req->seq,
          REQUEST_FROM(req));
    capn_free(&c);

    free_packet(pkt);
    FREEUP(pkt);

    return 0;
}

void init_reply_by_req(xce_reply_t* reply, xce_request_t* req)
{
    /* Zero the whole reply first.  Callers declare `xce_reply_t reply;` on the
     * stack and only populate the fields they need (status / data_which / data),
     * leaving the common trailer fields (redirect_host, redirect_port) untouched.
     * The generated encode_RequestReply reads redirect_host unconditionally
     * (gated only on != NULL), so a stale stack value there is dereferenced via
     * strlen() — a heap-use-after-free when the garbage aliases a freed capnp
     * segment.  Clearing here makes every reply path safe by construction. */
    memset(reply, 0, sizeof(*reply));
    reply->seq    = req->seq;
    reply->opcode = req->opcode;
}

/*
 * xce_free_request: free an xce_request_t and ALL owned sub-fields.
 *
 * The generated free_Request() handles the capnp-decoded fields but does
 * not know about raw_sha256/raw_bytes on upgrade_config_req_t (those are
 * extra fields populated by xce_decode() before capn_free()).  This wrapper
 * frees those extra fields first, then calls free_Request().
 *
 * Use this everywhere instead of calling free_Request() directly.
 */
void xce_free_request(xce_request_t *req)
{
    if (req == NULL)
        return;

    /* 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;
        free(ucr->raw_sha256);
        ucr->raw_sha256    = NULL;
        free(ucr->raw_bytes);
        ucr->raw_bytes     = NULL;
        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;
    }

    if (req->body_which == Request_body_walcatchupentrymsg
        && req->body.walcatchupentrymsg != NULL) {
        wal_catchup_entry_t *batch = req->body.walcatchupentrymsg;
        free(batch->raw_payload);
        batch->raw_payload     = NULL;
        batch->raw_payload_len = 0;
        free(batch->raw_batch_payloads);
        batch->raw_batch_payloads     = NULL;
        batch->raw_batch_payloads_len = 0;
        for (int i = 0; i < batch->n_entries; i++) {
            if (!batch->entries || !batch->entries[i])
                continue;
            free(batch->entries[i]->raw_payload);
            batch->entries[i]->raw_payload     = NULL;
            batch->entries[i]->raw_payload_len = 0;
        }
    }

    if (req->body_which == Request_body_snapshottransfermsg
        && req->body.snapshottransfermsg != NULL) {
        snapshot_transfer_t *chunk = req->body.snapshottransfermsg;
        free(chunk->raw_payload);
        chunk->raw_payload     = NULL;
        chunk->raw_payload_len = 0;
    }

    free_Request(req);
}

int xce_request(endpoint_t* ep, xce_request_t* req)
{
    int          rc;
    io_packet_t* pkt;

    pkt = (io_packet_t*)calloc(1, sizeof(*pkt));
    if (pkt == NULL) {
        return -ENOMEM;
    }

    req->seq  = endpoint_getseq(ep);
    req->from = ep->peer;

    rc = xce_encode(req, pkt);
    if (rc < 0) {
        free_packet(pkt);
        free(pkt);

        return rc;
    }

    int enq_rc = endpoint_enq(ep, &(pkt->link));
    return (enq_rc < 0) ? enq_rc : 0;
}

int xce_reply(endpoint_t* ep, xce_peerinfo_t* from, xce_reply_t* reply)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_reply,
        .body_which = Request_body_replymsg,
    };
    /* clang-format on */
    int          rc;
    io_packet_t* pkt;

    TRACE("%s seq %lu opcode %d status %d data which %d", __func__, reply->seq,
          reply->opcode, reply->status, reply->data_which);

    pkt = (io_packet_t*)calloc(1, sizeof(*pkt));
    if (pkt == NULL) {
        return -ENOMEM;
    }

    request.from          = from;
    request.body.replymsg = reply;

    request.seq = endpoint_getseq(ep);

    rc = xce_encode(&request, pkt);
    if (rc < 0) {
        free_packet(pkt);
        free(pkt);

        return rc;
    }

    int enq_rc = endpoint_enq(ep, &(pkt->link));
    return (enq_rc < 0) ? enq_rc : 0;
}

/*
 * xce_replconfigpush_req_raw: send OP_replConfigPush with raw byte arrays.
 *
 * Bypasses encode_ReplConfigPushRequest_ptr (which maps sha256/bytes through
 * xce_data_t — a cluster-data discriminated union that cannot carry raw bytes).
 * Instead, constructs the struct Request / struct ReplConfigPushRequest directly
 * using the capnp wire-format API (capn_new_list8 + capn_setv8), then calls
 * write_Request to serialise.
 */
int xce_replconfigpush_req_raw(endpoint_t *ep, xce_peerinfo_t *from,
                                uint64_t version_tag,
                                const uint8_t *sha256, size_t sha256_len,
                                const uint8_t *config_bytes, size_t config_len)
{
    struct capn               c;
    struct capn_segment      *cs;
    struct Request            rq;
    struct ReplConfigPushRequest push_req;
    ReplConfigPushRequest_ptr push_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(&push_req, 0, sizeof(push_req));

    push_req.versionTag = version_tag;

    /* sha256: write as a List(UInt8) — capn_list8 and capn_data share {capn_ptr p} */
    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);
        push_req.sha256.p = sha_list.p;
    }

    /* config bytes */
    if (config_bytes && config_len > 0) {
        capn_list8 bytes_list = capn_new_list8(cs, (int)config_len);
        capn_setv8(bytes_list, 0, config_bytes, (int)config_len);
        push_req.bytes.p = bytes_list.p;
    }

    push_ptr = new_ReplConfigPushRequest(cs);
    write_ReplConfigPushRequest(&push_req, push_ptr);

    rq.opcode     = OPCode_replConfigPush;
    rq.seq        = endpoint_getseq(ep);
    rq.body_which = Request_body_replconfigpushrequestmsg;
    rq.body.replconfigpushrequestmsg = push_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);

    int enq_rc = endpoint_enq(ep, &(pkt->link));
    return (enq_rc < 0) ? enq_rc : 0;
}

/* wal_entry_to_wire: fill a wire WalCatchupEntry from a C entry's header
 * fields and raw payload bytes.  Sub-entry nesting is one level deep —
 * elements written here never carry their own entries list. */
static void wal_entry_to_wire(struct capn_segment *cs,
                              struct WalCatchupEntry *w,
                              const wal_catchup_entry_t *e)
{
    memset(w, 0, sizeof(*w));
    w->lsn       = e->lsn;
    w->entryType = e->entry_type;
    /* ha-term-safety task 3.2: carry the entry's term stamp so the standby
     * intake persists the LEADER's recorded term (R1.3/R1.4), not a substitute. */
    w->term      = e->term;
    /* ha-term-safety task 3.4: carry the AppendEntries prev tuple + leaderCommit
     * PER SUB-ENTRY so a multi-entry catch-up batch is routable through the
     * library one entry at a time (each sub-entry's prev tuple differs; the
     * top-level container can hold only one). For a single-entry broadcast this
     * is redundant with the top-level metadata but harmless. */
    w->prevLogIndex = e->prev_log_index;
    w->prevLogTerm  = e->prev_log_term;
    w->leaderCommit = e->leader_commit;
    /* bug-1067: carry the streaming leader's CURRENT term per sub-entry (distinct
     * from `term` above, the entry's stored term). The standby intake uses this as
     * the AppendEntries header term so historical (entry_term < leader_term)
     * catch-up entries are not stale-rejected. */
    w->leaderCurrentTerm = e->leader_current_term;
    /* bug-847: do NOT write this sub-entry's payload as its own per-element
     * Data blob — that is exactly the shape that loses its pointer once a
     * multi-entry batch crosses a capnp segment boundary. The caller
     * (xce_wal_batch_encode_raw) instead carries every sub-entry's bytes
     * concatenated in ONE top-level batchPayloads blob; this scalar length
     * is all a sub-entry needs to find its slice on decode. */
    (void)cs;
    w->wirePayloadLen = (uint32_t)e->raw_payload_len;
}

/*
 * xce_wal_batch_encode_raw: encode an OPCode_walCatchupEntry batch into pkt.
 *
 * Bypasses encode_WalCatchupEntry (whose payload field routes through
 * xce_data_t — cannot carry raw bytes; same limitation as replConfigPush).
 * Builds the Request manually: a top-level WalCatchupEntry container
 * (lsn=0, entryType=0) whose entries list holds one element per batch
 * entry, each with lsn/entryType and payload written as capnp Data from
 * raw_payload/raw_payload_len.
 *
 * pkt is caller-allocated; on success it holds the serialized message
 * (caller owns; enqueue or free with free_packet).  Returns 0 or -errno.
 */
int xce_wal_batch_encode_raw(wal_catchup_entry_t **entries, int n_entries,
                              xce_peerinfo_t *from, uint64_t seq,
                              io_packet_t *pkt)
{
    struct capn          c;
    struct capn_segment *cs;
    struct Request       rq;
    struct WalCatchupEntry top;
    WalCatchupEntry_ptr  tp;
    Request_ptr          rp;
    int                  rc;

    /* n_entries == 0 is a valid AppendEntries HEARTBEAT (ha-term-safety task 3.2):
     * entries[0] still carries the AppendEntries metadata (prev tuple +
     * leader_commit) on the top-level container, with an empty entries list. */
    if (pkt == NULL || n_entries < 0 || entries == NULL)
        return -EINVAL;

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;

    memset(&rq,  0, sizeof(rq));
    memset(&top, 0, sizeof(top));

    /* ha-term-safety task 3.2: the top-level container carries the AppendEntries
     * metadata (prev tuple + leaderCommit) so a 0-entry heartbeat still conveys
     * UpdateTerm/logOk inputs. entries[0] is the metadata holder for both the
     * heartbeat (n_entries==0) and the single-entry (n_entries==1) AE built by
     * repl_send_append_entries. Legacy catch-up batches leave these zero. */
    top.term          = entries[0]->term;
    top.prevLogIndex  = entries[0]->prev_log_index;
    top.prevLogTerm   = entries[0]->prev_log_term;
    top.leaderCommit  = entries[0]->leader_commit;
    /* bug-1067: the AppendEntries header term (leader's CURRENT term), distinct
     * from top.term (entries[0]'s stored term). Mirrors the per-entry field. */
    top.leaderCurrentTerm = entries[0]->leader_current_term;

    /* Batch container: entries list carries the per-entry payload(s) (empty for
     * a heartbeat). Each sub-entry carries lsn/entryType/term/wirePayloadLen. */
    WalCatchupEntry_list lst = new_WalCatchupEntry_list(cs, n_entries);
    for (int i = 0; i < n_entries; i++) {
        struct WalCatchupEntry w;
        wal_entry_to_wire(cs, &w, entries[i]);
        set_WalCatchupEntry(&w, lst, i);
    }
    top.entries = lst;

    /* bug-847: carry every sub-entry's payload concatenated in ONE top-level
     * Data blob instead of one Data blob per sub-entry — a single blob does
     * not exhibit the cross-segment pointer loss that scattered per-element
     * Data blobs do in a multi-entry batch. Decode slices this back apart
     * per entry using each entry's own wirePayloadLen (set above). */
    {
        size_t total = 0;
        for (int i = 0; i < n_entries; i++)
            total += entries[i]->raw_payload_len;
        if (total > 0) {
            uint8_t *concat = (uint8_t *)malloc(total);
            if (!concat) {
                capn_free(&c);
                return -ENOMEM;
            }
            size_t off = 0;
            for (int i = 0; i < n_entries; i++) {
                if (entries[i]->raw_payload && entries[i]->raw_payload_len > 0) {
                    memcpy(concat + off, entries[i]->raw_payload,
                          entries[i]->raw_payload_len);
                    off += entries[i]->raw_payload_len;
                }
            }
            capn_list8 pl = capn_new_list8(cs, (int)total);
            capn_setv8(pl, 0, concat, (int)total);
            top.batchPayloads.p = pl.p;
            free(concat);
        }
    }

    tp = new_WalCatchupEntry(cs);
    write_WalCatchupEntry(&top, tp);

    rq.opcode     = OPCode_walCatchupEntry;
    rq.seq        = seq;
    rq.body_which = Request_body_walcatchupentrymsg;
    rq.body.walcatchupentrymsg = tp;
    encode_PeerInfo_ptr(cs, &rq.from, from);

    rp = new_Request(cs);
    write_Request(&rq, rp);
    capn_setp(capn_root(&c), 0, rp.p);

    pkt->sz = capn_size(&c);
    rc = alloc_packet(pkt);
    if (rc < 0) {
        capn_free(&c);
        return -errno;
    }

    capn_write_mem(&c, (uint8_t *)pkt->data, pkt->sz, 0);
    capn_free(&c);
    return 0;
}

int xce_wal_batch_req_raw(endpoint_t *ep, wal_catchup_entry_t **entries,
                           int n_entries)
{
    if (ep == NULL)
        return -EINVAL;

    io_packet_t *pkt = (io_packet_t *)calloc(1, sizeof(*pkt));
    if (!pkt)
        return -ENOMEM;

    int rc = xce_wal_batch_encode_raw(entries, n_entries, ep->peer,
                                       endpoint_getseq(ep), pkt);
    if (rc < 0) {
        free(pkt);
        return rc;
    }

    int enq_rc = endpoint_enq(ep, &(pkt->link));
    return (enq_rc < 0) ? enq_rc : 0;
}

/*
 * xce_snapshot_chunk_encode_raw: encode an OPCode_snapshotTransfer chunk
 * into pkt.  Envelope fields from *chunk; payload bytes written as capnp
 * Data from payload/payload_len (the legacy xce_data_t payload field is
 * ignored).  pkt is caller-allocated.  Returns 0 or -errno.
 */
int xce_snapshot_chunk_encode_raw(const snapshot_transfer_t *chunk,
                                   const uint8_t *payload, size_t payload_len,
                                   xce_peerinfo_t *from, uint64_t seq,
                                   io_packet_t *pkt)
{
    struct capn          c;
    struct capn_segment *cs;
    struct Request       rq;
    struct SnapshotTransfer st;
    SnapshotTransfer_ptr sp;
    Request_ptr          rp;
    int                  rc;

    if (pkt == NULL || chunk == NULL)
        return -EINVAL;

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;

    memset(&rq, 0, sizeof(rq));
    memset(&st, 0, sizeof(st));

    st.snapshotLsn    = chunk->snapshot_lsn;
    st.chunkIndex     = chunk->chunk_index;
    st.chunkOffset    = chunk->chunk_offset;
    st.totalSize      = chunk->total_size;
    st.isLast         = chunk->is_last ? 1 : 0;
    /* raft-ha-core-safety task 3.2: checkpoint's term at snapshotLsn, so the
     * receiver can reconcile its per-LSN term index at the boundary. */
    st.checkpointTerm = chunk->checkpoint_term;
    if (payload && payload_len > 0) {
        capn_list8 pl = capn_new_list8(cs, (int)payload_len);
        capn_setv8(pl, 0, payload, (int)payload_len);
        st.payload.p = pl.p;
    }

    sp = new_SnapshotTransfer(cs);
    write_SnapshotTransfer(&st, sp);

    rq.opcode     = OPCode_snapshotTransfer;
    rq.seq        = seq;
    rq.body_which = Request_body_snapshottransfermsg;
    rq.body.snapshottransfermsg = sp;
    encode_PeerInfo_ptr(cs, &rq.from, from);

    rp = new_Request(cs);
    write_Request(&rq, rp);
    capn_setp(capn_root(&c), 0, rp.p);

    pkt->sz = capn_size(&c);
    rc = alloc_packet(pkt);
    if (rc < 0) {
        capn_free(&c);
        return -errno;
    }

    capn_write_mem(&c, (uint8_t *)pkt->data, pkt->sz, 0);
    capn_free(&c);
    return 0;
}

int xce_snapshot_chunk_req_raw(endpoint_t *ep, const snapshot_transfer_t *chunk,
                                const uint8_t *payload, size_t payload_len)
{
    if (ep == NULL)
        return -EINVAL;

    io_packet_t *pkt = (io_packet_t *)calloc(1, sizeof(*pkt));
    if (!pkt)
        return -ENOMEM;

    int rc = xce_snapshot_chunk_encode_raw(chunk, payload, payload_len,
                                            ep->peer, endpoint_getseq(ep), pkt);
    if (rc < 0) {
        free(pkt);
        return rc;
    }

    int enq_rc = endpoint_enq(ep, &(pkt->link));
    return (enq_rc < 0) ? enq_rc : 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);

    int enq_rc = endpoint_enq(ep, &(pkt->link));
    return (enq_rc < 0) ? enq_rc : 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_XceData_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_XceData
     * which would corrupt via xce_data_t).  XceData_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: extract sha256 + bytes from a decoded
 * OP_upgradeConfig request.
 *
 * xce_decode() now extracts the raw List(UInt8) bytes from the capnp
 * segment before capn_free(), storing malloc'd copies in
 * req->body.upgradeconfigrequestmsg->raw_sha256 and ->raw_bytes.
 * This function returns shallow copies (pointers into the struct); the
 * caller must NOT free them — they are owned by the xce_request_t and
 * released by free_Request().
 *
 * Returns 0 on success, -EINVAL on bad args or missing fields.
 */
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;

    if (req->body_which != Request_body_upgradeconfigrequestmsg
        || req->body.upgradeconfigrequestmsg == NULL) {
        return -EINVAL;
    }

    const upgrade_config_req_t *ucr = req->body.upgradeconfigrequestmsg;
    if (ucr->raw_sha256 == NULL || ucr->raw_bytes == NULL
        || ucr->raw_bytes_len == 0) {
        return -EINVAL;
    }

    *out_sha256 = ucr->raw_sha256;
    *out_bytes  = ucr->raw_bytes;
    *out_len    = ucr->raw_bytes_len;
    return 0;
}

/*
 * 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
        = { .opcode = OPCode_nop, .body_which = Request_body_nomsg };

    if (ep == NULL) {
        return -EINVAL;
    }

    return xce_request(ep, &request);
}

/* xce_ha_heartbeat_req: send the HA heartbeat as an OP_nop request carrying a
 * HeartbeatMsg body with the primary's committed position. Clone of the nop
 * send path plus a heartbeatmsg body. Non-blocking, no reply awaited (the
 * standby's StateReport rides the normal nop reply). senderPriority is unused
 * (0). Generic xce_nop_req stays body-less. */
int xce_ha_heartbeat_req(endpoint_t* ep, uint64_t term, uint64_t committed_lsn,
                         uint64_t committed_config_tag)
{
    heartbeat_msg_t hb = {
        .sender_priority      = 0,
        .term                 = term,
        .committed_lsn        = committed_lsn,
        .committed_config_tag = committed_config_tag,
    };
    /* clang-format off */
    xce_request_t request = {
        .opcode     = OPCode_nop,
        .body_which = Request_body_heartbeatmsg,
        .body       = {
            .heartbeatmsg = &hb
        }
    };
    /* clang-format on */

    if (ep == NULL) {
        return -EINVAL;
    }

    return xce_request(ep, &request);
}

int xce_hello_req(endpoint_t* ep, char* msg)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode     = OPCode_hello,
        .body_which = Request_body_hellomsg,
        .body       = {
                  .hellomsg = msg
	}
    };
    /* clang-format on */

    if ((ep == NULL) || (msg == NULL)) {
        return -EINVAL;
    }
    return xce_request(ep, &request);
}

int xce_join_req(endpoint_t* ep, int role, hostattrs_t* attrs)
{
    /* clang-format off */
    xce_joindata_t joinmsg = {
        .role = role
    };
    xce_request_t request = {
        .opcode = OPCode_join,
        .body_which = Request_body_joinmsg,
        .body = {
            .joinmsg = &joinmsg 
        }
    };

    if ((role != HostRole_client) &&
        (role != HostRole_worker) &&
        (role != HostRole_apiserver) &&
        (role != HostRole_scheduler)) {
        return -EINVAL;
    }
    /* clang-format on */

    if (role == HostRole_worker) {
        if (attrs == NULL) {
            return -EINVAL;
        }

        joinmsg.data_which = JoinData_data_attrs;
        joinmsg.data.attrs = attrs;
    }
    return xce_request(ep, &request);
}

int xce_shutdown_req(endpoint_t* ep)
{
    int           rc;
    dataq_t*      outq;
    xce_request_t request
        = { .opcode = OPCode_shutdown, .body_which = Request_body_nomsg };

    if (ep == NULL) {
        return -EINVAL;
    }

    rc = xce_request(ep, &request);
    if (rc < 0) {
        return rc;
    }

    outq = &(ep->outq);

    while (!DATAQ_EMPTY(outq)) {
        sleep(1);
    }

    return 0;
}

int xce_pushconfig_req(endpoint_t* ep, xce_peerinfo_t* from,
                       xce_configdata_t* config)
{
    int rc = 0;
    /* clang-format off */
    xce_request_t request
        = { .opcode = OPCode_push, .body_which = Request_body_configpushmsg };
    /* clang-format on */

    if ((ep == NULL) || (config == NULL)) {
        return -EINVAL;
    }

    request.body.configpushmsg = config;
    rc                         = xce_request(ep, &request);

    return rc;
}

static int xce_push_req(endpoint_t* ep, xce_peerinfo_t* from, xce_data_t* data)
{
    int           rc = 0;
    xce_request_t request
        = { .opcode = OPCode_push, .body_which = Request_body_pushmsg };

    if ((ep == NULL) || (from == NULL) || (data == NULL)) {
        return -EINVAL;
    }

    request.body.pushmsg = data;

    rc = xce_request(ep, &request);
    return rc;
}

int xce_push_data_req(endpoint_t* ep, xce_peerinfo_t* from, xce_data_t* data,
                      xce_check_fn_t checkf, void* checkarg, int pktlimit)
{
    int        nrows = 0;
    int        pos   = 0;
    int        rc    = 0;
    xce_data_t data1;
    int        checkstatus = 0;

    if ((ep == NULL) || (from == NULL) || (data == NULL)) {
        return -EINVAL;
    }

    data1.table = data->table;
    switch (data->table) {
    case XceData_data_schedjobs:
    case XceData_data_jobs:
        nrows           = data->data.jobs->n_total;
        data1.data.jobs = (xce_jobs_data_t*)calloc(1, sizeof(xce_jobs_data_t));
        if (data1.data.jobs == NULL) {
            rc = -ENOMEM;
        }
        break;
    case XceData_data_schedworkers:
    case XceData_data_workers:
        nrows = data->data.workers->n_total;
        data1.data.workers
            = (xce_workers_data_t*)calloc(1, sizeof(xce_workers_data_t));
        if (data1.data.workers == NULL) {
            rc = -ENOMEM;
        }
        break;
    case XceData_data_resources:
        break;
    case XceData_data_resvals:
        break;
    default:
        rc = -EINVAL;
        break;
    }

    if (rc < 0) {
        return rc;
    }

    if (pktlimit <= 0) {
        pktlimit = nrows;
    }

    do {
        int nc = nrows - pos;

        if (nc > pktlimit) {
            nc = pktlimit;
        }

        if (checkf != NULL) {
            checkstatus = (*checkf)(checkarg);
            if (checkstatus < 0) {
                nc = 0;
            }
        }

        switch (data->table) {
        case XceData_data_schedjobs:
        case XceData_data_jobs:
            data1.data.jobs->n_total = data->data.jobs->n_total;
            data1.data.jobs->pos     = pos;
            data1.data.jobs->count   = nc;
            if (nc == 0) {
                data1.data.jobs->jobs = NULL;
            } else {
                data1.data.jobs->jobs = &(data->data.jobs->jobs[pos]);
            }
            break;
        case XceData_data_schedworkers:
        case XceData_data_workers:
            data1.data.workers->n_total = data->data.workers->n_total;
            data1.data.workers->pos     = pos;
            data1.data.workers->count   = nc;
            if (nc == 0) {
                data1.data.workers->workers = NULL;
            } else {
                data1.data.workers->workers
                    = &(data->data.workers->workers[pos]);
            }
            break;
        case XceData_data_resources:
            break;
        case XceData_data_resvals:
            break;
        default:
            break;
        }

        rc = xce_push_req(ep, from, &data1);

        if (rc < 0) {
            break;
        }

        pos += nc;

    } while ((pos < nrows) && (checkstatus >= 0));

    switch (data->table) {
    case XceData_data_schedjobs:
    case XceData_data_jobs:
        FREEUP(data1.data.jobs);
        break;
    case XceData_data_schedworkers:
    case XceData_data_workers:
        FREEUP(data1.data.workers);
        break;
    default:
        break;
    }

    return rc;
}

int xce_get_req(endpoint_t* ep, xce_get_t* get)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_get,
        .body_which = Request_body_getmsg,
        .body = {
            .getmsg = get
        }
    };
    /* clang-format on */

    return xce_request(ep, &request);
}

int xce_reply_get(endpoint_t* ep, xce_peerinfo_t* from, xce_reply_t* reply,
                  xce_check_fn_t checkf, void* checkarg, int pktlimit)
{
    int rc    = 0;
    int nrows = 0;
    int pos   = 0;
    /* clang-format off */
    xce_data_t getdata = {
        .table = reply->data.getdata->table
    };
    xce_reply_t reply1 = {
        .seq        = reply->seq,
        .opcode     = reply->opcode,
        .status     = reply->status,
        .data_which = reply->data_which,
        .data       = {
            .getdata = &getdata
        }
    };
    /* clang-format on */
    int         checkstatus = 0;
    xce_data_t* data        = reply->data.getdata;

    if ((ep == NULL) || (from == NULL) || (reply == NULL)) {
        return -EINVAL;
    }

    switch (getdata.table) {
    case XceData_data_schedjobs:
    case XceData_data_jobs:
        nrows = data->data.jobs->n_total;
        getdata.data.jobs
            = (xce_jobs_data_t*)calloc(1, sizeof(xce_jobs_data_t));
        break;
    case XceData_data_schedworkers:
    case XceData_data_workers:
        nrows = data->data.workers->n_total;
        getdata.data.workers
            = (xce_workers_data_t*)calloc(1, sizeof(xce_workers_data_t));
        break;
    case XceData_data_resources:
        nrows = data->data.resources->n_total;
        getdata.data.resources
            = (xce_resource_data_t*)calloc(1, sizeof(xce_resource_data_t));
        break;
    case XceData_data_resvals:
        nrows = data->data.resvals->n_total;
        getdata.data.resvals
            = (xce_resval_data_t*)calloc(1, sizeof(xce_resval_data_t));
        break;
    default:
        rc = -EINVAL;
        break;
    }

    if (rc < 0) {
        /* FIXME we need to reply no data */
        return rc;
    }

    if (pktlimit <= 0) {
        pktlimit = nrows;
    }

    do {
        int nc = nrows - pos;

        if (nc > pktlimit) {
            nc = pktlimit;
        }

        if (checkf != NULL) {
            checkstatus = (*checkf)(checkarg);

            if (checkstatus < 0) {
                nc = 0;
            }
        }

        switch (getdata.table) {
        case XceData_data_schedjobs:
        case XceData_data_jobs:
            getdata.data.jobs->n_total = data->data.jobs->n_total;
            getdata.data.jobs->pos     = pos;
            getdata.data.jobs->count   = nc;
            if (nc == 0) {
                getdata.data.jobs->jobs = NULL;
            } else {
                getdata.data.jobs->jobs = &(data->data.jobs->jobs[pos]);
            }
            break;
        case XceData_data_schedworkers:
        case XceData_data_workers:
            getdata.data.workers->n_total = data->data.workers->n_total;
            getdata.data.workers->pos     = pos;
            getdata.data.workers->count   = nc;
            if (nc == 0) {
                getdata.data.workers->workers = NULL;
            } else {
                getdata.data.workers->workers
                    = &(data->data.workers->workers[pos]);
            }
            break;
        case XceData_data_resources:
            getdata.data.resources->n_total = data->data.resources->n_total;
            getdata.data.resources->pos     = pos;
            getdata.data.workers->count     = nc;
            if (nc == 0) {
                getdata.data.resources->resources = NULL;
            } else {
                getdata.data.resources->resources
                    = &(data->data.resources->resources[pos]);
            }
            break;
        case XceData_data_resvals:
            getdata.data.resvals->n_total = data->data.resvals->n_total;
            getdata.data.resvals->pos     = pos;
            getdata.data.resvals->count   = nc;
            if (nc == 0) {
                getdata.data.resvals->vals = NULL;
            } else {
                getdata.data.resvals->vals = &(data->data.resvals->vals[pos]);
            }
            break;
        default:
            /* should never goes here */
            break;
        }

        rc = xce_reply(ep, from, &reply1);
        if (rc < 0) {
            break;
        }

        pos += nc;
    } while ((pos < nrows) && (checkstatus >= 0));

    switch (getdata.table) {
    case XceData_data_schedjobs:
    case XceData_data_jobs:
        FREEUP(getdata.data.jobs);
        break;
    case XceData_data_schedworkers:
    case XceData_data_workers:
        FREEUP(getdata.data.workers);
        break;
    case XceData_data_resources:
        FREEUP(getdata.data.resources);
        break;
    case XceData_data_resvals:
        FREEUP(getdata.data.resvals);
        break;
    default:
        break;
    }

    return rc;
}

int extract_data(dataset_t* dataset, xce_data_t* getdata)
{
    int ntotal;
    int rc    = 0;
    int pos   = 0;
    int count = 0;

    TRACE("%s ...", __func__);

    switch (getdata->table) {
    case XceData_data_schedjobs:
    case XceData_data_jobs:
        if (getdata->data.jobs->count <= 0) {
            return 1;
        }

        ntotal = getdata->data.jobs->n_total;
        pos    = getdata->data.jobs->pos;
        count  = getdata->data.jobs->count;

        if (dataset->rows == NULL) {
            dataset->rows  = (xce_job_t**)calloc(ntotal, sizeof(xce_job_t*));
            dataset->nrows = 0;
        }

        memcpy((void*)(((char*)dataset->rows) + sizeof(xce_job_t*) * pos),
               getdata->data.jobs->jobs, count * sizeof(xce_job_t*));
        memset(getdata->data.jobs->jobs, 0x0, count * sizeof(xce_job_t*));

        break;
    case XceData_data_schedworkers:
    case XceData_data_workers:
        if (getdata->data.workers->count <= 0) {
            return 1;
        }

        ntotal = getdata->data.workers->n_total;
        if (dataset->rows == NULL) {
            dataset->rows
                = (xce_worker_t**)calloc(ntotal, sizeof(xce_worker_t*));
            dataset->nrows = 0;
        }
        pos   = getdata->data.workers->pos;
        count = getdata->data.workers->count;

        memcpy((void*)(((char*)dataset->rows) + sizeof(xce_worker_t*) * pos),
               getdata->data.workers->workers, count * sizeof(xce_worker_t*));
        memset(getdata->data.workers->workers, 0x0,
               count * sizeof(xce_worker_t*));

        break;
    case XceData_data_resources:
        if (getdata->data.resources->count <= 0) {
            return 1;
        }

        ntotal = getdata->data.resources->n_total;
        if (dataset->rows == NULL) {
            dataset->rows
                = (xce_resource_t**)calloc(ntotal, sizeof(xce_resource_t*));
            dataset->nrows = 0;
        }
        pos   = getdata->data.resources->pos;
        count = getdata->data.resources->count;

        memcpy((void*)(((char*)dataset->rows) + sizeof(xce_resource_t*) * pos),
               getdata->data.resources->resources,
               count * sizeof(xce_resource_t*));

        memset(getdata->data.resources->resources, 0x0,
               count * sizeof(xce_resource_t*));
        break;
    case XceData_data_resvals:
        if (getdata->data.resvals->count <= 0) {
            return 1;
        }

        ntotal = getdata->data.resvals->n_total;
        if (dataset->rows == NULL) {
            dataset->rows
                = (xce_resval_t**)calloc(ntotal, sizeof(xce_resval_t*));
            dataset->nrows = 0;
        }

        pos   = getdata->data.resvals->pos;
        count = getdata->data.resvals->count;

        memcpy((void*)(((char*)dataset->rows) + sizeof(xce_resval_t*) * pos),
               getdata->data.resvals->vals, count * sizeof(xce_resval_t*));

        memset(getdata->data.resvals->vals, 0x0, count * sizeof(xce_resval_t*));
        break;
    case XceData_data_jobArray:
        /* job-array-single-wal-record task 5.3b: single, unchunked JobSubmit
         * descriptor (base jobid, array_size, one shared spec) -- unlike the
         * LIST-shaped cases above, no pagination/accumulation loop is needed:
         * xcesched_push_jobarray() (sched.c) and apiserver_push_jobarray()
         * (apiserver.c) always send exactly one complete descriptor per
         * commit. Transfer ownership of the decoded pointer straight into
         * dataset->rows (mirrors the LIST cases' ownership-transfer intent,
         * without an array-of-pointers layer since there is only one row)
         * and signal reassembly complete immediately -- returning here skips
         * the shared ntotal/pos-based tail below, which does not apply to a
         * single-shot payload. */
        if (getdata->data.job_array == NULL) {
            return -EINVAL;
        }

        dataset->rows            = (void*)getdata->data.job_array;
        dataset->nrows           = 1;
        getdata->data.job_array  = NULL;

        return 1;
    default:
        rc = -EINVAL;
        break;
    }

    if (rc < 0) {
        return rc;
    }

    dataset->nrows += count;
    if (dataset->nrows < ntotal) {
        return 0;
    }

    return 1;
}

int xce_jobdispatch_req(endpoint_t* ep, xce_job_dispatch_t* info)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_jobdispatch,
        .body_which = Request_body_dispatchmsg,
        .body = {
            .dispatchmsg = info 
        }
    };
    /* clang-format on */

    /* clang-format off */
    if ((ep == NULL) ||
        (info == NULL)) {
        return -EINVAL;
    }
    /* clang-format on */

    return xce_request(ep, &request);
}

int xce_notify_req(endpoint_t* ep, xce_notify_data_t* data)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_notify,
        .body_which = Request_body_notifymsg,
        .body = {
            .notifymsg = data
        }
    };
    /* clang-format on */

    if ((ep == NULL) || (data == NULL)) {
        return -EINVAL;
    }

    return xce_request(ep, &request);
}

int xce_jobsubmit_req(endpoint_t* ep, xce_job_submit_t* submit)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_jobsub,
        .body_which = Request_body_jobsubmsg,
        .body = {
            .jobsubmsg = submit
        }
    };
    /*clang-format on*/
    
    /* clang-format off */
    if ((ep == NULL) ||
        (submit == NULL)) {
        return -EINVAL;
    }
    /* clang-format on */

    return xce_request(ep, &request);
}

int xce_startjob_req(endpoint_t* ep, xce_job_exec_t* execinfo)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_startjob,
        .body_which = Request_body_jobstartmsg,
        .body = {
            .jobstartmsg = execinfo
        }
    };
    /* clang-format on */

    return xce_request(ep, &request);
}

int xce_clusterconf_req(endpoint_t* ep)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_clusterconf,
        .body_which = Request_body_nomsg
    };
    /* clang-format on */

    return xce_request(ep, &request);
}

int xce_ctrlstatus_req(endpoint_t* ep)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_ctrlstatus,
        .body_which = Request_body_nomsg
    };
    /* clang-format on */

    return xce_request(ep, &request);
}

int xce_update_req(endpoint_t* ep, xce_update_data_t* data)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_update,
        .body_which = Request_body_updatemsg,
        .body = {
            .updatemsg = data
        }
    };
    /* clang-format on */

    if ((ep == NULL) || (data == NULL)) {
        return -EINVAL;
    }

    return xce_request(ep, &request);
}

int xce_jobctrl_req(endpoint_t* ep, xce_jobctrl_t* ctrinfo)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_jobcontrol,
        .body_which = Request_body_jobctrlmsg,
        .body = {
            .jobctrlmsg = ctrinfo
        }
    };
    /* clang-format on */

    if ((ep == NULL) || (ctrinfo == NULL)) {
        return -EINVAL;
    }

    if (ctrinfo->jobid == 0) {
        return -EINVAL;
    }

    switch (ctrinfo->action) {
    case JobControl_u_siginfo:
        break;
    case JobControl_u_ckptoption:
        if (ctrinfo->u.ckptoption->method == NULL) {
            ctrinfo->u.ckptoption->method = "default";
        }
        break;
    default:
        return -EINVAL;
        break;
    }
    return xce_request(ep, &request);
}

int xce_queuectrl_req(endpoint_t* ep, xce_queuectrl_t* ctrlinfo)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_queuecontrol,
        .body_which = Request_body_queuectrlmsg,
        .body = {
            .queuectrlmsg = ctrlinfo
        }
    };
    /* clang-format on */

    if ((ep == NULL) || (ctrlinfo == NULL)) {
        return -EINVAL;
    }

    if ((ctrlinfo->queue == NULL) || (strlen(ctrlinfo->queue) == 0)) {
        return -EINVAL;
    }

    if (ctrlinfo->action == QUEUE_DO_NONE) {
        return 0;
    }

    if ((ctrlinfo->action < 0) || (ctrlinfo->action > QUEUE_ACTIVE)) {
        return -EINVAL;
    }

    return xce_request(ep, &request);
}

int xce_hostctrl_req(endpoint_t* ep, xce_hostctrl_t* ctrlinfo)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_hostcontrol,
        .body_which = Request_body_hostctrlmsg,
        .body = {
            .hostctrlmsg = ctrlinfo
        }
    };
    /* clang-format on */

    if ((ep == NULL) || (ctrlinfo == NULL)) {
        return -EINVAL;
    }

    if ((ctrlinfo->hostname == NULL) || (strlen(ctrlinfo->hostname) == 0)) {
        return -EINVAL;
    }

    if (ctrlinfo->action == HOST_DO_NONE) {
        return 0;
    }

    if (ctrlinfo->action != HOST_OPEN && ctrlinfo->action != HOST_CLOSE) {
        return -EINVAL;
    }

    return xce_request(ep, &request);
}

int xce_runtask_req(endpoint_t* ep, xce_taskrun_t* runinfo)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_runtasks,
        .body_which = Request_body_runtaskmsg,
        .body = {
            .runtaskmsg = runinfo
        }
    };
    /* clang-format on */

    if ((ep == NULL) || (runinfo == NULL)) {
        return -EINVAL;
    }

    if (runinfo->nhosts <= 0) {
        return -EINVAL;
    }

    return xce_request(ep, &request);
}

int xce_set_pend_reason_req(endpoint_t* ep, xce_preason_set_t* reasons)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_setpreason,
        .body_which = Request_body_preasonmsg,
        .body = {
            .preasonmsg = reasons
        }
    };
    /* clang-format on */

    if ((ep == NULL) || (reasons == NULL)) {
        return -EINVAL;
    }

    return xce_request(ep, &request);
}

int xce_clear_reason_req(endpoint_t* ep, xce_clear_reason_t* reason)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_clrpreason,
        .body_which = Request_body_clrpreasonmsg,
        .body = {
            .clrpreasonmsg = reason
        }
    };
    /* clang-format on */

    if ((ep == NULL) || (reason == NULL)) {
        return -EINVAL;
    }

    return xce_request(ep, &request);
}

int xce_fetchfile_req(endpoint_t* ep, xce_fetchfile_t* msg)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode     = OPCode_fetchfile,
        .body_which = Request_body_fetchfilemsg,
        .body = {
            .fetchfilemsg = msg
        }
    };
    /* clang-format on */

    if ((ep == NULL) || (msg == NULL)) {
        return -EINVAL;
    }

    if (msg->filepath == NULL || msg->filepath[0] == '\0') {
        return -EINVAL;
    }

    return xce_request(ep, &request);
}

/* ── Client-Side Transparent Retry tunables (Req 6.1, 6.2) ─────────────────
 *
 * XCE_RETRY_DEFAULTS provides safe production defaults for the failover retry
 * loop that lives in the libxce client layer (client.c / client_failover.c).
 * The orphaned no-op xce_request_with_retry() primitive and its
 * xce_retry_send_hook test seam were removed; they are superseded by the
 * xcecli_* failover loop.
 */
const xce_retry_config_t XCE_RETRY_DEFAULTS = {
    .max_retries        = 20,
    .initial_retry_ms   = 50,
    .max_retry_ms       = 5000,
    .backoff_multiplier = 2.0,
};