diff --git a/src/controller/config_repl.c b/src/controller/config_repl.c index 7ddb296..09ae4d8 100644 --- a/src/controller/config_repl.c +++ b/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 ──────────────────────── */ diff --git a/src/include/request.h b/src/include/request.h index a48b4b1..1f98301 100644 --- a/src/include/request.h +++ b/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 diff --git a/src/lib/request.c b/src/lib/request.c index 0409ff5..fb67c34 100644 --- a/src/lib/request.c +++ b/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