feat: implement push_config_to_standbys() — parallel fan-out via xce_replconfigpush_req_raw() using capn_new_list8 for raw bytes

src/controller/config_repl.c

@@ -773,39 +773,109 @@ static void send_upgrade_error(endpoint_t *ep, xce_request_t *req,
 
 /* ── push_config_to_standbys ─────────────────────── */
 
-/*
- * push_config_to_standbys: fan-out new config blob to connected standbys.
- *
- * Task 11 dependency: the OPCode_replConfigPush wire protocol requires the
- * repl_config_push_req_t.sha256 and .bytes fields to carry raw byte data.
- * In the current typemap these fields are typed as xce_data_t* (a discriminated
- * cluster-data union), not a raw byte buffer.  The capnp Data-type shim that
- * maps them to plain byte arrays is wired in Task 11.
- *
- * Until Task 11 lands, standbys catch up automatically via handle_config_request()
- * on the next connection.  The primary ACKs acked_count=0; the admin tool knows
- * to wait for standbys to reconnect and pull.
- */
+typedef struct {
+    endpoint_t      *ep;
+    xce_peerinfo_t  *from;
+    uint64_t         new_tag;
+    const uint8_t   *bytes;
+    size_t           blen;
+    const uint8_t   *sha256;
+    atomic_int      *stop;
+    int              acked;
+} push_ctx_t;
+
+static void *push_thread(void *arg)
+{
+    push_ctx_t *ctx = (push_ctx_t *)arg;
+    if (atomic_load(ctx->stop))
+        return NULL;
+
+    int rc = xce_replconfigpush_req_raw(
+        ctx->ep, ctx->from,
+        ctx->new_tag,
+        ctx->sha256, 32,
+        ctx->bytes,  ctx->blen);
+
+    if (rc == 0)
+        ctx->acked = 1;
+    else
+        WARN("push_config_to_standbys: push to endpoint %p failed: %d", ctx->ep, rc);
+
+    return NULL;
+}
+
 int push_config_to_standbys(cluster_t *c, const uint8_t *bytes, size_t blen,
                             const uint8_t sha256[32], uint64_t new_tag)
 {
-    (void)bytes;
-    (void)blen;
-    (void)sha256;
-    (void)new_tag;
+    if (!c->ha.standby_endpoints)
+        return 0;
 
-    int n = 0;
-    if (c->ha.standby_endpoints) {
-        pthread_mutex_lock(&c->ha.lock);
-        n = c->ha.standby_endpoints->count;
+    /* Snapshot connected standbys under lock */
+    pthread_mutex_lock(&c->ha.lock);
+    int n = c->ha.standby_endpoints->count;
+    if (n == 0) {
         pthread_mutex_unlock(&c->ha.lock);
+        return 0;
+    }
+    endpoint_t **eps = malloc((size_t)n * sizeof(endpoint_t *));
+    if (!eps) {
+        pthread_mutex_unlock(&c->ha.lock);
+        return 0;
+    }
+    for (int i = 0; i < n; i++)
+        eps[i] = (endpoint_t *)zarray_get(c->ha.standby_endpoints, i);
+    pthread_mutex_unlock(&c->ha.lock);
+
+    atomic_int stop = 0;
+    push_ctx_t *ctxs    = calloc((size_t)n, sizeof(push_ctx_t));
+    pthread_t  *threads = malloc((size_t)n * sizeof(pthread_t));
+    if (!ctxs || !threads) {
+        free(ctxs); free(threads); free(eps);
+        return 0;
     }
 
-    if (n > 0)
-        WARN("push_config_to_standbys: %d standby(s) connected but push"
-             " deferred to Task 11 (typemap byte shims not yet wired);"
-             " standbys will pull on reconnect", n);
-    return 0;
+    for (int i = 0; i < n; i++) {
+        ctxs[i].ep      = eps[i];
+        ctxs[i].from    = c->myhost;
+        ctxs[i].new_tag = new_tag;
+        ctxs[i].bytes   = bytes;
+        ctxs[i].blen    = blen;
+        ctxs[i].sha256  = sha256;
+        ctxs[i].stop    = &stop;
+        ctxs[i].acked   = 0;
+        pthread_create(&threads[i], NULL, push_thread, &ctxs[i]);
+    }
+
+    /* Aggregate wait: CONF_PUSH_TIMEOUT_MS total, not per-standby */
+    struct timespec deadline;
+    clock_gettime(CLOCK_MONOTONIC, &deadline);
+    deadline.tv_sec  += CONF_PUSH_TIMEOUT_MS / 1000;
+    deadline.tv_nsec += (CONF_PUSH_TIMEOUT_MS % 1000) * 1000000L;
+    if (deadline.tv_nsec >= 1000000000L) {
+        deadline.tv_sec++;
+        deadline.tv_nsec -= 1000000000L;
+    }
+    while (1) {
+        struct timespec now;
+        clock_gettime(CLOCK_MONOTONIC, &now);
+        if (now.tv_sec > deadline.tv_sec ||
+            (now.tv_sec == deadline.tv_sec && now.tv_nsec >= deadline.tv_nsec))
+            break;
+        struct timespec sl = { .tv_sec = 0, .tv_nsec = 5 * 1000000L };
+        nanosleep(&sl, NULL);
+    }
+
+    atomic_store(&stop, 1);
+    int acked = 0;
+    for (int i = 0; i < n; i++) {
+        pthread_join(threads[i], NULL);
+        if (ctxs[i].acked) acked++;
+    }
+
+    INFO("Config pushed to %d/%d standbys", acked, n);
+
+    free(ctxs); free(threads); free(eps);
+    return acked;
 }
 
 /* ── handle_upgrade_config ──────────────────────── */

src/include/request.h

@@ -59,4 +59,16 @@ extern int xce_runtask_req(endpoint_t*, xce_taskrun_t*);
 extern int xce_set_pend_reason_req(endpoint_t*, xce_preason_set_t*);
 extern int xce_clear_reason_req(endpoint_t*, xce_clear_reason_t*);
 
+/*
+ * xce_replconfigpush_req_raw: send OP_replConfigPush carrying raw byte arrays.
+ * Bypasses the encode_ReplConfigPushRequest_ptr path (which maps sha256/bytes
+ * through xce_data_t — wrong type for raw bytes) by constructing the capnp
+ * message directly via capn_new_list8 + capn_setv8 + write_Request.
+ */
+extern 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);
+
 #endif

src/lib/request.c

@@ -142,6 +142,88 @@ int xce_reply(endpoint_t* ep, xce_peerinfo_t* from, xce_reply_t* reply)
     return 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);
+
+    endpoint_enq(ep, &(pkt->link));
+    return 0;
+}
+
 int xce_nop_req(endpoint_t* ep)
 {
     xce_request_t request