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
This commit is contained in:
parent
3cf82721d6
commit
0c4a0760d1
2 changed files with 208 additions and 1 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue