From 910b43cd376b6155d7b1633efce698bb9aa222f4 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Fri, 25 Sep 2026 14:55:04 +0000 Subject: [PATCH] fix(controller): never close() the primary channel's raw fd (bug-1865) handle_ctrlhello stored the netio endpoint's fd (primary_ctrl_fd = ep->fd) and ha_on_primary_ctrl_fd_rdhup close()d it in three branches. netio owns that socket and closes it on endpoint teardown, so by the time the liveness poll fired the RDHUP the number could be reused -- in a chaotic soak by the standby's WAL segment: HA's close() shut the WAL, fdatasync failed 2305 times (EBADF, then EINVAL once a socket took the number), durable_lsn stuck, and one round's jobsub failed EIO. handle_ctrlhello now also holds a reference on the endpoint (primary_ctrl_ep), which keeps its socket open and its fd number ours. ha_drop_primary_ctrl_conn() replaces the raw closes: shutdown() the still-owned socket so netio sees the hang-up and tears it down itself, then release the reference. A bare fd with no endpoint is only forgotten. Test test_rdhup_never_closes_an_fd_netio_already_released: real ctrlhello -> netio teardown -> file opened on the freed number -> RDHUP; the file was closed before, survives now. test_rdhup_closes_and_nulls_fd updated to the new contract (peer reads EOF, fields cleared). Controller suite 5/5. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/controller/config_repl.c | 5 ++ src/controller/ha.c | 37 +++++++---- src/controller/ha.h | 15 ++++- .../testsuite/test_ha_connection_reuse.c | 65 +++++++++++++++++++ src/controller/testsuite/test_ha_failover.c | 27 +++++--- 5 files changed, 127 insertions(+), 22 deletions(-) diff --git a/src/controller/config_repl.c b/src/controller/config_repl.c index a9a646dd..224a79f6 100644 --- a/src/controller/config_repl.c +++ b/src/controller/config_repl.c @@ -2190,6 +2190,11 @@ int handle_ctrlhello(cluster_t *c, endpoint_t *ep, xce_request_t *req) memory_order_acquire) == CTRL_STANDBY && c->ha.primary_ctrl_fd == -1) { c->ha.primary_ctrl_fd = ep->fd; + /* bug-1865: hold the endpoint so netio cannot close (and the kernel + * reuse) this fd while HA treats it as the primary channel; released + * by ha_drop_primary_ctrl_conn */ + endpoint_ref(ep); + c->ha.primary_ctrl_ep = ep; /* bug-1701: ha->primary_host/primary_port (aliased into a redirect * reply by fill_redirect_reply(), join.c:27) were never written by diff --git a/src/controller/ha.c b/src/controller/ha.c index 27556654..01d2a82d 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -2889,6 +2889,7 @@ int ha_init(ha_state_t *ha, const ha_config_t *cfg, int my_priority) ha->jitter_seed = 0; ha->standby_endpoints = NULL; ha->primary_ctrl_fd = -1; + ha->primary_ctrl_ep = NULL; /* election-iomgr-refactor Task 2.1: explicit zero-init for documentation clarity * (memset above already zeroes these, but being explicit is better documentation) */ @@ -8645,6 +8646,27 @@ void ha_begin_failover(ha_state_t *ha, cluster_t *c) ha_start_prevote_async(ha, c); } +/* bug-1865: drop the standby's channel to the primary. primary_ctrl_fd is a + * netio endpoint's socket: netio owns it and closes it on endpoint teardown, + * after which the fd number can be reused (it was, by the WAL segment) -- + * so it is never close()d here. The endpoint is held with a reference + * (handle_ctrlhello), which keeps the socket open and its number ours until + * the reference is released: shutdown() it so netio sees the hang-up and + * tears the connection down itself, then release the reference. */ +void ha_drop_primary_ctrl_conn(ha_state_t *ha) +{ + endpoint_t *ep = (endpoint_t *)ha->primary_ctrl_ep; + + if (ep != NULL) { + if (ep->fd >= 0) { + (void)shutdown(ep->fd, SHUT_RDWR); + } + ha->primary_ctrl_ep = NULL; + endpoint_unref(ep); + } + ha->primary_ctrl_fd = -1; +} + void ha_on_primary_ctrl_fd_rdhup(ha_state_t *ha, cluster_t *c) { if (!ha) return; @@ -8665,10 +8687,7 @@ void ha_on_primary_ctrl_fd_rdhup(ha_state_t *ha, cluster_t *c) ctrl_role_t cur = (ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire); if (cur == CTRL_ELECTING || ha->raft.pre_vote_in_progress) { - if (ha->primary_ctrl_fd >= 0) { - close(ha->primary_ctrl_fd); - ha->primary_ctrl_fd = -1; - } + ha_drop_primary_ctrl_conn(ha); INFO("HA: primary_ctrl_fd RDHUP but an election is already in progress " "(role=%d, prevote=%d) — dropping the RDHUP, not restarting " "failover (task 5.6: don't preempt the in-progress election / " @@ -8696,10 +8715,7 @@ void ha_on_primary_ctrl_fd_rdhup(ha_state_t *ha, cluster_t *c) memory_order_acquire); uint64_t cooldown_ms = 2u * (uint64_t)ha->config.election_timeout_ms; if (last != 0 && now_ms >= last && (now_ms - last) < cooldown_ms) { - if (ha->primary_ctrl_fd >= 0) { - close(ha->primary_ctrl_fd); - ha->primary_ctrl_fd = -1; - } + ha_drop_primary_ctrl_conn(ha); INFO("HA: primary_ctrl_fd RDHUP within failover cooldown (%lums since " "last trigger, cooldown=%lums) — debouncing, not restarting " "failover (5.10b: a sustained RDHUP cadence must not ladder the " @@ -8728,10 +8744,7 @@ void ha_on_primary_ctrl_fd_rdhup(ha_state_t *ha, cluster_t *c) /* Close and null the primary control fd. Connection-teardown bookkeeping * stays at the trigger regardless of the PreVote outcome (ha-prevote task * 4.2). */ - if (ha->primary_ctrl_fd >= 0) { - close(ha->primary_ctrl_fd); - ha->primary_ctrl_fd = -1; - } + ha_drop_primary_ctrl_conn(ha); /* raft-ha-core-safety task 5.2: SPURIOUS-FAILOVER SUPPRESSION. * diff --git a/src/controller/ha.h b/src/controller/ha.h index 262a547e..67a13481 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -178,6 +178,12 @@ typedef struct { uint64_t voted_for_term; /* Req 9: term of last granted vote */ char voted_for_name[XCE_MAXNAME_LEN]; /* Req 9: candidate of last granted vote */ int primary_ctrl_fd; /* Req 6: persistent channel fd (-1 = none) */ + /* bug-1865: the netio endpoint primary_ctrl_fd belongs to, held with a + * reference (handle_ctrlhello) so its socket -- which netio owns and + * closes -- cannot be closed and its fd number reused while HA still + * treats it as the primary channel. Dropped via ha_drop_primary_ctrl_conn, + * never by close(primary_ctrl_fd). NULL = none. */ + struct endpoint_t* primary_ctrl_ep; bool defer_config_fetch; /* Req 3: tag=0 node defers fetch to post-election */ void * _Atomic elect_ctx; /* Req 2: ha_elect_ctx_t* during election; NULL otherwise */ _Atomic uint64_t last_known_committed_lsn; /* Req 5: from primary heartbeats */ @@ -1605,12 +1611,17 @@ ctrl_role_t ha_wait_for_heartbeat_or_timeout(ha_state_t *ha); /* ── Task 9.4: EPOLLRDHUP Failover and Heartbeat Timeout (Req 6.2–6.4, 6.8) ─ * ha_on_primary_ctrl_fd_rdhup: EPOLLRDHUP event handler. - * Closes primary_ctrl_fd, transitions to CTRL_ELECTING, truncates WAL to - * last_known_committed_lsn, fires ha_start_election_async (via cluster when non-NULL). + * Drops the primary channel (ha_drop_primary_ctrl_conn), transitions to + * CTRL_ELECTING, truncates WAL to last_known_committed_lsn, fires + * ha_start_election_async (via cluster when non-NULL). * Pass c == NULL to skip persist/election (unit test mode). */ void ha_on_primary_ctrl_fd_rdhup(ha_state_t *ha, cluster_t *c); +/* bug-1865: shut down + release the standby's primary channel. Never + * close()s primary_ctrl_fd -- netio owns that socket. */ +void ha_drop_primary_ctrl_conn(ha_state_t *ha); + /* * ha_arm_heartbeat_timeout_vtask: start/re-arm the heartbeat watchdog thread. * Fires ha_on_primary_ctrl_fd_rdhup after election_timeout_ms + rand_jitter of diff --git a/src/controller/testsuite/test_ha_connection_reuse.c b/src/controller/testsuite/test_ha_connection_reuse.c index 76cf0b59..7a321927 100644 --- a/src/controller/testsuite/test_ha_connection_reuse.c +++ b/src/controller/testsuite/test_ha_connection_reuse.c @@ -12,6 +12,9 @@ */ #define _GNU_SOURCE #include +#include +#include +#include #include #include #include @@ -427,6 +430,67 @@ static void test_handle_ctrlhello_sets_primary_ctrl_fd(void) printf("test_handle_ctrlhello_sets_primary_ctrl_fd: PASS\n"); } +/* bug-1865: handle_ctrlhello kept a copy of the netio endpoint's raw fd and + * ha_on_primary_ctrl_fd_rdhup close()d it -- but netio owns that socket and + * closes it on endpoint teardown. By the time the liveness poll fired the + * RDHUP, the number had been reused by the standby's WAL segment, and HA's + * close() shut the WAL instead (fdatasync EBADF/EINVAL, durable_lsn stuck). + * Drive the real path: ctrlhello records the heap endpoint, netio drops its + * own reference (teardown), a file is opened, then RDHUP -- the file must + * survive. */ +extern void ha_on_primary_ctrl_fd_rdhup(ha_state_t *ha, cluster_t *c); + +static void test_rdhup_never_closes_an_fd_netio_already_released(void) +{ + cluster_t c; + myconf_t conf; + ctrl_hello_req_t hello; + xce_request_t req; + endpoint_t* ep; + int sv[2]; + int file_fd; + + memset(&c, 0, sizeof(c)); + ha_init(&c.ha, &g_cfg, 0); + c.ha.primary_ctrl_fd = -1; + atomic_store_explicit(&c.ha.role, (int32_t)CTRL_STANDBY, + memory_order_release); + memset(&conf, 0, sizeof(conf)); + c.myconf = &conf; + + TEST_ASSERT_EQUAL_INT(0, socketpair(AF_UNIX, SOCK_STREAM, 0, sv)); + ep = endpoint_get(); /* heap ep, netio's own reference */ + TEST_ASSERT_NOT_NULL(ep); + ep->fd = sv[0]; + + memset(&hello, 0, sizeof(hello)); + hello.name = "primary"; + memset(&req, 0, sizeof(req)); + req.body_which = Request_body_ctrlhellomsg; + req.body.ctrlhellomsg = &hello; + (void)handle_ctrlhello(&c, ep, &req); + TEST_ASSERT_EQUAL_INT(sv[0], c.ha.primary_ctrl_fd); + + /* netio tears the connection down (drops its reference) */ + endpoint_unref(ep); + + /* the standby then opens its WAL segment: lowest free fd */ + file_fd = open("/dev/null", O_RDONLY); + TEST_ASSERT_TRUE(file_fd >= 0); + + ha_on_primary_ctrl_fd_rdhup(&c.ha, NULL); + + TEST_ASSERT_TRUE_MESSAGE(fcntl(file_fd, F_GETFD) != -1, + "RDHUP must not close an fd that netio released -- here it had been " + "reused by an unrelated file (the WAL in production)"); + TEST_ASSERT_EQUAL_INT(-1, c.ha.primary_ctrl_fd); + + close(file_fd); + close(sv[1]); + ha_fini(&c.ha); + printf("test_rdhup_never_closes_an_fd_netio_already_released: PASS\n"); +} + /* ── Test 8b (bug-1701): handle_ctrlhello populates primary_host/port from * the sender's name matched against our own static peers_cfg candidate list. * This is the local-lookup fix that unblocks bug-1700's fill_redirect_reply() @@ -631,6 +695,7 @@ void test_ha_connection_reuse(void) /* Req 3.1 — handle_ctrlhello designates primary control channel */ RUN_TEST(test_handle_ctrlhello_sets_primary_ctrl_fd); + RUN_TEST(test_rdhup_never_closes_an_fd_netio_already_released); /* bug-1701 — handle_ctrlhello populates primary_host/primary_port via a * local peers_cfg lookup keyed by req->from->name */ diff --git a/src/controller/testsuite/test_ha_failover.c b/src/controller/testsuite/test_ha_failover.c index 065409fe..b5e6164f 100644 --- a/src/controller/testsuite/test_ha_failover.c +++ b/src/controller/testsuite/test_ha_failover.c @@ -25,6 +25,7 @@ #include #include "ha.h" +#include "netio.h" #include "xce/cluster_def.h" /* full cluster_t layout for task-5.2 tests */ #include "myconf.h" /* myconf_t for task-5.2 tests */ @@ -91,28 +92,38 @@ static ha_config_t g_fast_cfg = { .election_priority_step_ms = 500, }; -/* ── Test 1: RDHUP closes fd and sets primary_ctrl_fd = -1 ─────────────── */ +/* ── Test 1: RDHUP drops the primary channel and sets primary_ctrl_fd = -1 ── + * bug-1865: the channel is a netio endpoint HA holds a reference on; RDHUP + * shuts its socket down (the peer sees EOF) and releases the reference -- + * it never close()s the raw fd, which netio owns. */ static void test_rdhup_closes_and_nulls_fd(void) { - ha_state_t ha; + ha_state_t ha; + endpoint_t* ep; + char b; + TEST_ASSERT_EQUAL_INT(0, ha_init(&ha, &g_cfg, 1)); - /* Set up a socketpair as the "primary_ctrl_fd" */ + /* Set up a socketpair as the primary channel, held like handle_ctrlhello */ int sv[2]; TEST_ASSERT_EQUAL_INT(0, socketpair(AF_UNIX, SOCK_STREAM, 0, sv)); + ep = endpoint_get(); + TEST_ASSERT_NOT_NULL(ep); + ep->fd = sv[0]; + endpoint_ref(ep); ha.primary_ctrl_fd = sv[0]; + ha.primary_ctrl_ep = ep; /* Fire RDHUP with NULL cluster (unit test mode — skips persist/election) */ ha_on_primary_ctrl_fd_rdhup(&ha, NULL); - /* fd must be closed and set to -1 */ + /* channel dropped: fields cleared, peer sees the hang-up */ TEST_ASSERT_EQUAL_INT(-1, ha.primary_ctrl_fd); - /* Verify sv[0] is really closed by trying to write to it */ - ssize_t n = write(sv[0], "x", 1); - TEST_ASSERT_LESS_THAN_INT(0, (int)n); - TEST_ASSERT_EQUAL_INT(EBADF, errno); + TEST_ASSERT_NULL(ha.primary_ctrl_ep); + TEST_ASSERT_EQUAL_INT(0, (int)read(sv[1], &b, 1)); + endpoint_unref(ep); /* netio's own reference: closes sv[0] */ close(sv[1]); ha_fini(&ha); printf("test_rdhup_closes_and_nulls_fd: PASS\n");