chore: baseline snapshot — prior-session work + ha-consistency-invariants spec family

Pre-existing uncommitted work: ha-repl-hardening payload data plane (7.x),
heartbeat liveness fix (Req 1.3), connection-reuse ASAN fixes, doc cleanup.
New: ha-consistency-invariants umbrella (requirements/design/research) and
its four implementation sub-specs (specs generated via batch, cross-spec
review applied), steering roadmap + product.md Raft+CP decision.

src/lib/delay.c

@@ -1,5 +1,4 @@
 #include <errno.h>
-#include <signal.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>

src/lib/meson.build

@@ -173,6 +173,7 @@ libxce_test_src = ['testsuite/test.c',
                    'testsuite/test_libxce_retry.c',
                    'testsuite/test_netio_on_connected.c',
                    'testsuite/test_raft.c',
+                   'testsuite/test_wal_wire.c',
                    'raft/raft_sim.c',
                   ]
 

src/lib/netio.c

@@ -4,6 +4,7 @@
 #include <netdb.h>
 #include <openssl/err.h>
 #include <openssl/ssl.h>
+#include <signal.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
@@ -949,6 +950,11 @@ SSL_CTX* create_ssl_context(char* cert_file, char* key_file)
 
 int iomgr_init(iomgr_t* iomgr)
 {
+    /* Ignore SIGPIPE — writing to a closed peer socket returns EPIPE
+     * instead of killing the daemon. All daemons call iomgr_init before
+     * any socket I/O, so this protects controller, worker, and scheduler. */
+    signal(SIGPIPE, SIG_IGN);
+
     mylock_init(&(iomgr->lock));
     iomgr->epoll_fd = epoll_create1(0);
     if (iomgr->epoll_fd < 0) {

src/lib/request.c

@@ -56,6 +56,35 @@ int xce_encode(xce_request_t* req, io_packet_t* pkt)
     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(Data_ptr 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;
@@ -158,6 +187,47 @@ int xce_decode(io_packet_t* pkt, xce_request_t* req)
         }
     }
 
+    /* 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);
+
+        /* Batch sub-entries */
+        WalCatchupEntry_list wlist = WalCatchupEntry_get_entries(b_ptr);
+        int wn = (wlist.p.type != CAPN_NULL) ? capn_len(wlist) : 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);
+            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);
@@ -210,6 +280,29 @@ void xce_free_request(xce_request_t *req)
         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;
+        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);
 }
 
@@ -356,6 +449,197 @@ int xce_replconfigpush_req_raw(endpoint_t *ep, xce_peerinfo_t *from,
     return 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;
+    if (e->raw_payload && e->raw_payload_len > 0) {
+        capn_list8 pl = capn_new_list8(cs, (int)e->raw_payload_len);
+        capn_setv8(pl, 0, e->raw_payload, (int)e->raw_payload_len);
+        w->payload.p = pl.p;
+    }
+}
+
+/*
+ * 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;
+
+    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));
+
+    /* Batch container: header fields zero, entries list carries the batch */
+    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;
+
+    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;
+    }
+
+    endpoint_enq(ep, &(pkt->link));
+    return 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;
+    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;
+    }
+
+    endpoint_enq(ep, &(pkt->link));
+    return 0;
+}
+
 /*
  * xce_replconfig_reply_raw: primary sends OP_replConfig reply with real
  * config bytes via an endpoint (uses iomgr queueing).

src/lib/testsuite/test.c

@@ -37,6 +37,7 @@ extern void test_cluster_parse(void);
 extern void test_libxce_retry(void);
 extern void test_netio_on_connected(void);
 extern void test_raft(void);
+extern void test_wal_wire(void);
 
 void setUp(void) { return; }
 
@@ -143,6 +144,7 @@ int main(int argc, char* argv[])
     TEST(test_libxce_retry);
     TEST(test_netio_on_connected);
     TEST(test_raft);
+    TEST(test_wal_wire);
 
     return UNITY_END();
 }