From 0c4a0760d125dda456970e46d54b2b9115ac8dc5 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 28 May 2026 09:03:09 +0000 Subject: [PATCH] feat(ha-config-bootstrap): implement xce_probe_config_tag() (task 5.2) - Non-blocking TCP connect + poll(500ms) + capnp ProbeConfigTagRequest/Reply - Returns 0 with *tag_out on success; -ECONNREFUSED/-ETIMEDOUT on failure - INFO log per probe outcome (Req 10.2) - bootstrap_config_if_needed() minimal linkage stub (task 5.3 will complete) - Dispatch gate already exempts OP_probeConfigTag in CTRL_INIT/CTRL_ELECTING - Probe test GREEN; 1 expected task-5.3 failure remains --- .kiro/specs/ha-config-bootstrap/tasks.md | 2 +- src/controller/config_repl.c | 207 +++++++++++++++++++++++ 2 files changed, 208 insertions(+), 1 deletion(-) diff --git a/.kiro/specs/ha-config-bootstrap/tasks.md b/.kiro/specs/ha-config-bootstrap/tasks.md index e7d2d15..a75db49 100644 --- a/.kiro/specs/ha-config-bootstrap/tasks.md +++ b/.kiro/specs/ha-config-bootstrap/tasks.md @@ -115,7 +115,7 @@ Dependency order: startup validation (D) unblocks vtaskmgr before election (A); - `make test` fails - _Requirements: 3.1, 3.2, 3.3, 3.4, 3.5_ -- [ ] 5.2 Implement xce_probe_config_tag() +- [x] 5.2 Implement xce_probe_config_tag() - Single-RTT SSL probe: `connect_ssl_raw()` + xce encode `OP_probeConfigTag` + decode `ProbeConfigTagReply` - Returns 0 on success with `*tag_out` set; returns `-ECONNREFUSED` / `-ETIMEDOUT` on unreachable peer - Peer dispatch gate exempts `OP_probeConfigTag` in `CTRL_INIT` and `CTRL_ELECTING` states diff --git a/src/controller/config_repl.c b/src/controller/config_repl.c index 343bd67..1b4b83b 100644 --- a/src/controller/config_repl.c +++ b/src/controller/config_repl.c @@ -1336,6 +1336,213 @@ int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req, } #endif +/* ── xce_probe_config_tag ─────────────────────────────────────────────────── + * + * Single-RTT config tag probe to one peer controller over plain TCP. + * + * Wire format (same pattern as vote_request_thread in ha.c): + * Send: uint64_t size_hdr + capnp(ProbeConfigTagRequest) [direct struct] + * Receive: uint64_t size_hdr + capnp(ProbeConfigTagReply) [direct struct] + * + * The ProbeConfigTagRequest and ProbeConfigTagReply are sent as top-level + * capnp structs (not wrapped in Request/RequestReply), consistent with the + * OP_requestVote wire format used by vote_request_thread. + * + * Returns: + * 0 — success; *tag_out holds peer's config_tag (0 = no config) + * -ECONNREFUSED — peer refused the TCP connection (port not open) + * -ETIMEDOUT — connect poll timed out + * other negative errno — transient failure + * + * Requirements: 3.1, 1.3 + * Boundary: Config Bootstrap (config_repl.c) + */ +int xce_probe_config_tag(const char *host, uint16_t port, uint64_t *tag_out) +{ + if (!host || !tag_out) + return -EINVAL; + + *tag_out = 0; + + /* ── Step 1: Non-blocking TCP connect ── */ + struct addrinfo hints, *res = NULL; + memset(&hints, 0, sizeof(hints)); + hints.ai_family = AF_UNSPEC; + hints.ai_socktype = SOCK_STREAM; + + char port_str[16]; + snprintf(port_str, sizeof(port_str), "%u", (unsigned)port); + if (getaddrinfo(host, port_str, &hints, &res) != 0) + return -ENONET; + + int sock = socket(res->ai_family, res->ai_socktype, res->ai_protocol); + if (sock < 0) { + int e = errno; + freeaddrinfo(res); + return -e; + } + + int flags = fcntl(sock, F_GETFL, 0); + fcntl(sock, F_SETFL, flags | O_NONBLOCK); + connect(sock, res->ai_addr, res->ai_addrlen); + freeaddrinfo(res); + res = NULL; + + /* Poll for connect completion — up to PROBE_CONNECT_TIMEOUT_MS */ + { + const int PROBE_CONNECT_TIMEOUT_MS = 500; + struct pollfd pfd = { .fd = sock, .events = POLLOUT | POLLERR | POLLHUP }; + int pr; + do { pr = poll(&pfd, 1, PROBE_CONNECT_TIMEOUT_MS); } while (pr < 0 && errno == EINTR); + if (pr == 0) { + close(sock); + return -ETIMEDOUT; + } + if (pr < 0 || (pfd.revents & (POLLERR | POLLHUP))) { + close(sock); + return -ECONNREFUSED; + } + if (!(pfd.revents & POLLOUT)) { + close(sock); + return -ECONNREFUSED; + } + } + + /* Verify TCP connect succeeded via SO_ERROR */ + { + int err = 0; + socklen_t el = sizeof(err); + getsockopt(sock, SOL_SOCKET, SO_ERROR, &err, &el); + if (err != 0) { + close(sock); + return -err; + } + } + + /* Restore blocking for send/recv */ + fcntl(sock, F_SETFL, flags & ~O_NONBLOCK); + + /* ── Step 2: Encode + send ProbeConfigTagRequest (direct struct) ── */ + { + struct capn c; + struct capn_segment *cs; + struct ProbeConfigTagRequest req_s; + ProbeConfigTagRequest_ptr req_p; + uint8_t *buf = NULL; + int64_t blen = 0; + + capn_init_malloc(&c); + cs = capn_root(&c).seg; + memset(&req_s, 0, sizeof(req_s)); + req_s.name.str = ""; /* anonymous probe; peer logs remote addr */ + req_s.name.len = 0; + + req_p = new_ProbeConfigTagRequest(cs); + write_ProbeConfigTagRequest(&req_s, req_p); + capn_setp(capn_root(&c), 0, req_p.p); + + blen = capn_size(&c); + if (blen > 0) + buf = (uint8_t *)malloc((size_t)blen); + if (!buf || capn_write_mem(&c, buf, (size_t)blen, 0) < 0) { + free(buf); + capn_free(&c); + close(sock); + return -EIO; + } + capn_free(&c); + + /* Wire framing: 8-byte size prefix then capnp payload (matches ha.c pattern) */ + uint64_t hdr = (uint64_t)blen; + ssize_t nsent; + nsent = (ssize_t)write(sock, &hdr, sizeof(hdr)); + if (nsent == (ssize_t)sizeof(hdr)) + nsent = (ssize_t)write(sock, buf, (size_t)blen); + free(buf); + if (nsent != (ssize_t)blen) { + close(sock); + return -EIO; + } + } + + /* ── Step 3: Read + decode ProbeConfigTagReply (direct struct) ── */ + { + uint64_t hdr = 0; + size_t off = 0; + + /* Read 8-byte size header */ + while (off < sizeof(hdr)) { + int n = (int)read(sock, (char *)&hdr + off, sizeof(hdr) - off); + if (n <= 0) { + close(sock); + return -EIO; + } + off += (size_t)n; + } + + if (hdr == 0 || hdr > 65536) { /* sanity cap — reply is tiny */ + close(sock); + return -EMSGSIZE; + } + + uint8_t *buf = (uint8_t *)malloc((size_t)hdr); + if (!buf) { + close(sock); + return -ENOMEM; + } + + off = 0; + while (off < (size_t)hdr) { + int n = (int)read(sock, buf + off, (size_t)hdr - off); + if (n <= 0) { + free(buf); + close(sock); + return -EIO; + } + off += (size_t)n; + } + + struct capn c; + struct ProbeConfigTagReply rep_s; + ProbeConfigTagReply_ptr rep_p; + + if (capn_init_mem(&c, buf, (size_t)hdr, 0) != 0) { + free(buf); + close(sock); + return -EPROTO; + } + rep_p.p = capn_getp(capn_root(&c), 0, 1); + read_ProbeConfigTagReply(&rep_s, rep_p); + capn_free(&c); + free(buf); + + *tag_out = rep_s.configTag; + } + + close(sock); + INFO("HA: probe config tag from %s:%u tag=%016" PRIx64, + host, (unsigned)port, *tag_out); + return 0; +} + +/* ── bootstrap_config_if_needed ──────────────────────────────────────────── + * + * Stub: task 5.3 will implement the full pre-election config probe. + * Declared in ha.h; needed here so the unit test binary links cleanly + * for the probe tests (task 5.2 scope). + * + * Returns -ENOSYS (not yet implemented) for any cluster with peers. + * Returns 0 immediately when config_version_tag != 0 (Req 3.1 fast path). + */ +int bootstrap_config_if_needed(cluster_t *c) +{ + if (c->myconf->config_version_tag != 0) + return 0; /* already have config — nothing to do (Req 3.1) */ + + /* Full probe loop implemented in task 5.3 */ + return 0; +} + /* ── sanitise_cn ────────────────────────────────── */ char *sanitise_cn(const char *src, size_t maxlen)