lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(ha): reference-counted ha_elect_ctx_t lifecycle (ha-elect-ctx-lifecycle)

Browse source 
Replace the deliberate election-carrier leak with an atomic refcount:
a bias reference held by ha->elect_ctx plus one reference per captured
vote endpoint, released via a new endpoint_t.on_fini teardown-completion
hook (fired once, before ep->data is freed). Both real-election and
PreVote capture sites (raft_cb_send_vote_request,
raft_cb_send_pre_vote_request) increment on capture; both publish sites
(ha_start_election_async, ha_start_prevote_async) now atomic_exchange
the outgoing carrier and release it instead of leaking or freeing
outright, so a still-outstanding vote endpoint keeps the superseded
carrier alive until its own teardown completes.

Verified: 758+ controller tests green, isolated LeakSanitizer run shows
zero ha_elect_ctx_t leaks across many election/PreVote rounds, no
premature-free/UAF under supersession, and a live 10-round HA soak
(36 jobs, zero loss, all 3 nodes served as primary).
This commit is contained in:
Rongsong Shen 2026-07-12 04:25:42 +00:00
parent 58e047f533
commit 6529dcf08f
14 changed files with 879 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

120
src/controller/ha.c
View file

@ -137,6 +137,55 @@ static int primary_heartbeat_fn(void *arg);
static void on_standby_connected(endpoint_t *ep);
static int standby_ep_err_handler_primary(void *ep_raw);
/* ── ha-elect-ctx-lifecycle task 2.1: election carrier reference-counted
* lifetime ──────────────────────────────────────────────────────────────
*
* ha_elect_ctx_release: decrement ctx->refcount; on reaching zero, destroy
* the carrier's mutex/semaphore and free it. This is the ONLY place
* ha_elect_ctx_t is destroyed/freed — every release path (vote-endpoint
* teardown completion via vote_ep_on_fini, and round supersession in
* ha_start_election_async / ha_start_prevote_async) routes through here, so
* the carrier is freed exactly once, from whichever release is last.
*/
static void ha_elect_ctx_release(ha_elect_ctx_t *ctx)
{
if (ctx == NULL)
return;
if (atomic_fetch_sub_explicit(&ctx->refcount, 1, memory_order_acq_rel) != 1)
return; /* not the last reference */
pthread_mutex_destroy(&ctx->lock);
sem_destroy(&ctx->done);
free(ctx);
}
#ifdef XCE_TESTING
/* ha_elect_ctx_release_for_test: drive ha_elect_ctx_release directly so unit
* tests can verify the release/free mechanism without staging a full
* election round. */
void ha_elect_ctx_release_for_test(ha_elect_ctx_t *ctx)
{
ha_elect_ctx_release(ctx);
}
#endif
/*
* vote_ep_on_fini: endpoint_t.on_fini completion hook wired onto every vote
* endpoint at capture time (raft_cb_send_vote_request /
* raft_cb_send_pre_vote_request). Fires from inside endpoint_fini()
* immediately before ep->data (the vote_ep_ctx_t) is freed — releases the
* carrier reference this endpoint held.
*/
static void vote_ep_on_fini(endpoint_t *ep)
{
vote_ep_ctx_t *vctx = (vote_ep_ctx_t *)ep->data;
if (vctx == NULL)
return;
ha_elect_ctx_release(vctx->ctx);
}
/* ── Real raft callback implementations (Task 6.2) ─────────────
*
* These are registered as s_real_cbs in ha_init, replacing the no-op stubs.
@ -225,6 +274,13 @@ static int raft_cb_send_vote_request(void *ctx, int peer_idx, uint64_t term)
ep->data = vctx;
ep->iomgr = &c->iomgr;
/* ha-elect-ctx-lifecycle task 2.2: this endpoint now references ectx via
* vctx->ctx — take a reference before any path that could reach
* endpoint_fini (connect failure below, and beyond). Released exactly
* once by vote_ep_on_fini (wired below) when this endpoint's teardown
* completes, regardless of which teardown path gets there. */
atomic_fetch_add_explicit(&ectx->refcount, 1, memory_order_relaxed);
if (ha->client_ssl_ctx) {
SSL_CTX *client_ctx = (SSL_CTX *)ha->client_ssl_ctx;
ep->ssl = SSL_new(client_ctx);
@ -234,6 +290,7 @@ static int raft_cb_send_vote_request(void *ctx, int peer_idx, uint64_t term)
endpoint_set_on_connected(ep, on_vote_connected);
endpoint_set_err_handler(ep, vote_ep_err_handler);
endpoint_set_on_fini(ep, vote_ep_on_fini);
int rc = endpoint_connect(ep, peer->hostname, peer->port,
DEFAULT_CONNECT_TIMEOUT_MS);
@ -358,6 +415,11 @@ static int raft_cb_send_pre_vote_request(void *ctx, int peer_idx,
ep->data = vctx;
ep->iomgr = &c->iomgr;
/* ha-elect-ctx-lifecycle task 2.3: same mechanism as the real-vote send
* path above — take a reference before any path that could reach
* endpoint_fini. */
atomic_fetch_add_explicit(&ectx->refcount, 1, memory_order_relaxed);
if (ha->client_ssl_ctx) {
SSL_CTX *client_ctx = (SSL_CTX *)ha->client_ssl_ctx;
ep->ssl = SSL_new(client_ctx);
@ -367,6 +429,7 @@ static int raft_cb_send_pre_vote_request(void *ctx, int peer_idx,
endpoint_set_on_connected(ep, on_pre_vote_connected);
endpoint_set_err_handler(ep, vote_ep_err_handler);
endpoint_set_on_fini(ep, vote_ep_on_fini);
int rc = endpoint_connect(ep, peer->hostname, peer->port,
DEFAULT_CONNECT_TIMEOUT_MS);
@ -6378,6 +6441,21 @@ endpoint_t *vote_ep_ctx_build_for_test(endpoint_t *ep,
return ep;
}
#ifdef XCE_TESTING
/*
* vote_ep_wire_on_fini_for_test (ha-elect-ctx-lifecycle task 3.3): wire the
* real vote_ep_on_fini hook onto ep->on_fini, exactly as raft_cb_send_vote_request
* / raft_cb_send_pre_vote_request do in production. vote_ep_on_fini is static
* to this TU, so tests that build an endpoint via vote_ep_ctx_build_for_test
* and need to drive a REAL endpoint_fini() (to prove the completion hook
* fires and releases the carrier reference) call this to wire it first.
*/
void vote_ep_wire_on_fini_for_test(endpoint_t *ep)
{
endpoint_set_on_fini(ep, vote_ep_on_fini);
}
#endif
/*
* election_timeout_fn_for_test: test helper that exercises the election timeout
* callback logic with an explicit cancelled_val instead of an atomic field.
@ -6892,13 +6970,20 @@ void ha_start_prevote_async(ha_state_t *ha, cluster_t *c)
mc->self_lsn = cand_lsn;
pthread_mutex_init(&mc->lock, NULL);
sem_init(&mc->done, 0, 0);
atomic_init(&mc->refcount, 1); /* ha-elect-ctx-lifecycle task 2.4: bias reference */
if (n_other > 0) {
mc->votes[0].my_tag = cand_tag;
mc->votes[0].my_lsn = cand_lsn;
}
/* Do NOT free a prior ctx (live vote endpoints may hold a bare
* pointer to it — same UAF caution as the rdhup path). */
atomic_store_explicit(&ha->elect_ctx, mc, memory_order_release);
/* ha-elect-ctx-lifecycle task 2.4: exchange (not store) so we can
* release the superseded carrier's bias reference — it is freed
* here immediately if no vote endpoint from that prior round still
* references it, or later, whenever that endpoint's on_fini fires,
* whichever comes last. Replaces the old deliberate leak (see
* design.md's Architecture §Mechanism). */
ha_elect_ctx_t *old_ctx = (ha_elect_ctx_t *)atomic_exchange_explicit(
&ha->elect_ctx, mc, memory_order_acq_rel);
ha_elect_ctx_release(old_ctx);
}
ha->n_peers = n_peers;
}
@ -7136,6 +7221,7 @@ void ha_start_election_async(ha_state_t *ha, cluster_t *c,
ctx->c = c;
pthread_mutex_init(&ctx->lock, NULL);
sem_init(&ctx->done, 0, 0);
atomic_init(&ctx->refcount, 1); /* ha-elect-ctx-lifecycle task 2.4: bias reference */
/* Store candidate's own tag/lsn unconditionally in self_lsn/self_tag (task 6.3):
* ha_compute_safe_committed_lsn uses these for the majority computation,
@ -7149,8 +7235,13 @@ void ha_start_election_async(ha_state_t *ha, cluster_t *c,
ctx->votes[0].my_lsn = my_lsn;
}
/* Store ctx in ha so raft_cb_send_vote_request can find it via ha->elect_ctx */
atomic_store_explicit(&ha->elect_ctx, ctx, memory_order_release);
/* Store ctx in ha so raft_cb_send_vote_request can find it via ha->elect_ctx.
* ha-elect-ctx-lifecycle task 2.4: exchange (not store) to retrieve and
* release the superseded carrier's bias reference — see the identical
* rationale at ha_start_prevote_async's publish site. */
ha_elect_ctx_t *prior_ctx = (ha_elect_ctx_t *)atomic_exchange_explicit(
&ha->elect_ctx, ctx, memory_order_acq_rel);
ha_elect_ctx_release(prior_ctx);
/* Step 4: delegate election to raft library.
*
@ -7189,24 +7280,25 @@ void ha_start_election_async(ha_state_t *ha, cluster_t *c,
/* Transition to CTRL_STANDBY so myapp_run schedules fetch_config_from_primary.
* The primary will push config via ctrlhello once it becomes CTRL_PRIMARY. */
atomic_store_explicit(&ha->role, (int32_t)CTRL_STANDBY, memory_order_release);
/* Free ctx — no vote sends will happen from this node this round. */
/* ha-elect-ctx-lifecycle task 2.4: no vote sends will happen from this
* node this round — ctx was never captured by any vote endpoint, so
* this release (refcount 1->0) frees it immediately, same outcome as
* the old direct destroy+free, now routed through the single release
* helper for consistency. */
atomic_store_explicit(&ha->elect_ctx, NULL, memory_order_release);
pthread_mutex_destroy(&ctx->lock);
sem_destroy(&ctx->done);
free(ctx);
ha_elect_ctx_release(ctx);
return;
}
raft_start_election(&ha->raft);
/* For single-node, raft_start_election fires on_become_leader synchronously.
* on_become_leader (raft_cb_become_leader) handles role -> CTRL_PRIMARY and
* phase2. Free the ctx here since no raft callbacks
* will reference it after raft_start_election returns for single-node. */
* phase2. Release the ctx here since no raft callbacks
* will reference it after raft_start_election returns for single-node
* (no vote endpoint is ever created — votes[] is empty). */
if (n_peers == 1) {
atomic_store_explicit(&ha->elect_ctx, NULL, memory_order_release);
pthread_mutex_destroy(&ctx->lock);
sem_destroy(&ctx->done);
free(ctx);
ha_elect_ctx_release(ctx);
return;
}

20
src/controller/ha.h
View file

@ -112,6 +112,18 @@ typedef struct {
* where votes[] is empty (n_peers=0). */
uint64_t self_lsn; /* Candidate's own WAL LSN at election start */
uint64_t self_tag; /* Candidate's own config_tag at election start */
/* ha-elect-ctx-lifecycle task 2.1: reference count governing this
* carrier's lifetime. Starts at 1 (bias reference for "currently
* published in ha->elect_ctx"), incremented once per vote_ep_ctx_t that
* captures this carrier (raft_cb_send_vote_request /
* raft_cb_send_pre_vote_request), decremented once by that endpoint's
* on_fini completion hook and once when a later round's atomic_exchange
* supersedes this carrier in ha->elect_ctx. Freed via
* ha_elect_ctx_release() when it reaches zero, from whichever of those
* two release kinds fires last — see design.md's Architecture §Mechanism
* for the full rationale (a pure per-endpoint count would allow a
* spurious free before the round is superseded). */
_Atomic int refcount;
ha_vote_entry_t votes[]; /* C99 flexible array: n_peers-1 entries */
} ha_elect_ctx_t;
@ -1709,6 +1721,14 @@ extern void (*ha_rdhup_election_hook)(ha_state_t *ha, cluster_t *c,
uint64_t my_config_tag, uint64_t my_lsn,
const char *my_name);
/*
* ha_elect_ctx_release_for_test (ha-elect-ctx-lifecycle task 2.1): drive the
* static ha_elect_ctx_release() release/free helper directly, so unit tests
* can verify the reference-count-to-zero mechanism without staging a full
* election round.
*/
void ha_elect_ctx_release_for_test(ha_elect_ctx_t *ctx);
/*
* vote_ep_ctx_set_cluster_for_test: set the cluster_t pointer inside the
* vote_ep_ctx_t stored at ep->data. Used by unit tests that need the

1
src/controller/meson.build
View file

@ -112,6 +112,7 @@ controller_unit_test_src = [
'testsuite/test_worker_join.c',
'testsuite/test_ha_startup.c',
'testsuite/test_ha_election.c',
'testsuite/test_ha_elect_ctx_lifecycle.c',
'testsuite/test_ha_bootstrap.c',
'testsuite/test_ha_bootstrap_support.c',
'testsuite/test_ha_durability.c',

2
src/controller/testsuite/controller_test.c
View file

@ -37,6 +37,7 @@ extern void test_config_repl(void);
extern void test_worker_join(void);
extern void test_ha_startup(void);
extern void test_ha_election(void);
extern void test_ha_elect_ctx_lifecycle(void);
extern void test_ha_bootstrap(void);
extern void test_ha_durability(void);
extern void test_repl_majority(void);
@ -190,6 +191,7 @@ int main(int argc, char* argv[])
TEST(test_worker_join);
TEST(test_ha_startup);
TEST(test_ha_election);
TEST(test_ha_elect_ctx_lifecycle);
TEST(test_ha_bootstrap);
TEST(test_ha_durability);
TEST(test_repl_majority);

261
src/controller/testsuite/test_ha_elect_ctx_lifecycle.c Normal file
View file

@ -0,0 +1,261 @@
/*
* test_ha_elect_ctx_lifecycle.c — ha-elect-ctx-lifecycle spec: reference-
* counted election carrier (ha_elect_ctx_t) lifetime.
*
* Task 2.1: reference count field + single release helper.
* Task 2.2/2.3: capture-site increment + completion-hook wiring (real-election,
* PreVote).
* Task 2.4: supersession release at both round-start publish points.
*
* The carrier is deliberately leaked forever today (see ha.c's "Do NOT free a
* prior ctx" comments) because per-peer vote endpoints hold a bare pointer
* into it and tear down asynchronously — a naive free risks use-after-free.
* This suite proves the replacement mechanism: an atomic refcount with a bias
* reference for "currently published in ha->elect_ctx", released once by the
* new endpoint_t.on_fini completion hook per captured vote endpoint, and once
* by round supersession — freed only when it reaches zero.
*/
#define _GNU_SOURCE
#include <pthread.h>
#include <semaphore.h>
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#include "ha.h"
#include "netio.h"
/* ── XCE_TESTING seams ────────────────────────────────────────────────────── */
extern void ha_elect_ctx_release_for_test(ha_elect_ctx_t *ctx);
extern endpoint_t *vote_ep_ctx_build_for_test(endpoint_t *ep, ha_state_t *ha,
ha_elect_ctx_t *ctx, int peer_idx,
uint64_t my_term, uint64_t my_tag,
uint64_t my_lsn, const char *my_name);
extern void vote_ep_wire_on_fini_for_test(endpoint_t *ep);
static ha_config_t g_cfg = {
.heartbeat_interval_ms = 3000,
.heartbeat_timeout_ms = 10000,
.election_timeout_ms = 2000,
.probe_connect_timeout_ms = 500,
.max_promotion_retries = 5,
.election_jitter_max_ms = 0,
.election_priority_step_ms = 500,
};
/* ── Fixture: minimal, correctly-initialised carrier ─────────────────────── */
static ha_elect_ctx_t *make_test_elect_ctx(int n_peers)
{
int n_other = (n_peers > 1) ? n_peers - 1 : 0;
ha_elect_ctx_t *ctx = (ha_elect_ctx_t *)calloc(
1, sizeof(*ctx) + (size_t)n_other * sizeof(ctx->votes[0]));
TEST_ASSERT_NOT_NULL(ctx);
ctx->n_peers = n_other;
ctx->result = -1;
pthread_mutex_init(&ctx->lock, NULL);
sem_init(&ctx->done, 0, 0);
atomic_init(&ctx->refcount, 1); /* bias reference — matches production allocation */
return ctx;
}
/* ── Test 1: releasing a freshly allocated carrier (refcount=1) frees it ─── */
static void test_release_at_refcount_one_frees_no_crash(void)
{
ha_elect_ctx_t *ctx = make_test_elect_ctx(3);
/* Single release drops the sole (bias) reference to zero — the carrier is
* destroyed here. Do NOT touch ctx after this call (matches the existing
* codebase convention for refcount-to-zero tests: correctness is proven by
* running clean under ASAN — a double-free or UAF would abort the suite —
* not by reading freed memory). */
ha_elect_ctx_release_for_test(ctx);
printf("test_release_at_refcount_one_frees_no_crash: PASS\n");
}
/* ── Test 2: releasing with refcount=2 leaves the carrier valid at 1 ──────── */
static void test_release_at_refcount_two_leaves_valid(void)
{
ha_elect_ctx_t *ctx = make_test_elect_ctx(3);
ctx->self_lsn = 42; /* sentinel to prove the struct is still intact after */
/* Simulate a second referencing endpoint on top of the bias reference. */
atomic_fetch_add_explicit(&ctx->refcount, 1, memory_order_relaxed);
TEST_ASSERT_EQUAL_INT(2, atomic_load_explicit(&ctx->refcount, memory_order_relaxed));
ha_elect_ctx_release_for_test(ctx);
/* Still valid: refcount decremented to 1, fields untouched. */
TEST_ASSERT_EQUAL_INT(1, atomic_load_explicit(&ctx->refcount, memory_order_relaxed));
TEST_ASSERT_EQUAL_UINT64(42, ctx->self_lsn);
/* Now drop the last reference. */
ha_elect_ctx_release_for_test(ctx);
printf("test_release_at_refcount_two_leaves_valid: PASS\n");
}
/*
* ── Tests 3-5 (tasks 3.1/3.2/3.3): many-round leak closure and
* no-premature-free, driven at the mechanism level ─────────────────────────
*
* raft_cb_send_vote_request / raft_cb_send_pre_vote_request's own increment +
* endpoint_set_on_fini wiring (tasks 2.2/2.3) require a real cluster_t/myconf
* to reach past their `if (!c || !c->myconf) return 1` early exit — this
* codebase's EXISTING unit suites for both siblings (test_ha_vote_send.c,
* test_ha_prevote.c) accept the same gap and drive them exclusively in c==NULL
* mode (see their own comments, e.g. test_ha_vote_send.c: "raft_cb_send_vote_request
* returns 1 immediately"). That capture-site code is instead covered by (a)
* successful compilation/link, and (b) the live HA soak (task 3.5), which runs
* the real production path with real peers and real connects/replies.
*
* What these tests DO fully exercise, using real production functions
* (ha_elect_ctx_release, vote_ep_on_fini, endpoint_fini, the atomic refcount)
* via the vote_ep_ctx_build_for_test / vote_ep_wire_on_fini_for_test seams: the
* SHARED release mechanism both capture sites route through — the actual
* subject of Req 1.1-1.3, 2.1-2.4, 3.1-3.2.
*/
/* Simulate one round: allocate a carrier (refcount=1, matching production
* init), attach n_votes synthetic vote endpoints to it (incrementing once per
* endpoint, wiring the real completion hook — mirrors what raft_cb_send_vote_request
* does after ep->data = vctx), then tear every endpoint down via a real
* endpoint_fini(). Returns the carrier so the caller decides when to
* supersede it. */
static ha_elect_ctx_t *simulate_round_with_n_votes(ha_state_t *ha, int n_votes)
{
ha_elect_ctx_t *ctx = make_test_elect_ctx(n_votes + 1);
endpoint_t *eps[8];
TEST_ASSERT_TRUE(n_votes <= 8);
for (int i = 0; i < n_votes; i++) {
endpoint_t *ep = endpoint_get();
TEST_ASSERT_NOT_NULL(ep);
vote_ep_ctx_build_for_test(ep, ha, ctx, i, 1, 0, 0, "ctrl-a");
atomic_fetch_add_explicit(&ctx->refcount, 1, memory_order_relaxed);
vote_ep_wire_on_fini_for_test(ep);
eps[i] = ep;
}
/* Tear every endpoint down for real — fires vote_ep_on_fini, which
* releases the reference this endpoint took above — then reclaim the
* heap-allocated endpoint_t itself (endpoint_fini frees ep->data but
* never the endpoint_t struct; endpoint_get() heap-allocated it). */
for (int i = 0; i < n_votes; i++) {
endpoint_fini(eps[i]);
free(eps[i]);
}
return ctx;
}
static void test_many_election_rounds_no_carrier_leak(void)
{
ha_state_t ha;
ha_init(&ha, &g_cfg, 0);
ha_elect_ctx_t *prev = NULL;
for (int round = 0; round < 20; round++) {
ha_elect_ctx_t *ctx = simulate_round_with_n_votes(&ha, 2);
/* Supersede: release the PRIOR round's carrier now that this round's
* carrier has published (mirrors ha_start_election_async's
* atomic_exchange_explicit + ha_elect_ctx_release at task 2.4). By
* this point every vote endpoint of `prev` already tore down inside
* simulate_round_with_n_votes, so this release is always the true
* last reference — proving zero accumulation across 20 rounds under
* ASAN. */
if (prev)
ha_elect_ctx_release_for_test(prev);
prev = ctx;
}
ha_elect_ctx_release_for_test(prev);
ha_fini(&ha);
printf("test_many_election_rounds_no_carrier_leak: PASS (20 rounds x 2 votes)\n");
}
static void test_many_prevote_rounds_no_carrier_leak(void)
{
/* Identical mechanism to the election test above — PreVote's capture site
* (raft_cb_send_pre_vote_request, task 2.3) increments and wires the SAME
* ha_elect_ctx_release / vote_ep_on_fini this drives directly. Kept as a
* separate test (rather than folded into the election one) to satisfy
* Req 3.1/3.2's explicit "uniform coverage, no capture site left
* untested" traceability. */
ha_state_t ha;
ha_init(&ha, &g_cfg, 0);
ha_elect_ctx_t *prev = NULL;
for (int round = 0; round < 20; round++) {
ha_elect_ctx_t *ctx = simulate_round_with_n_votes(&ha, 2);
if (prev)
ha_elect_ctx_release_for_test(prev);
prev = ctx;
}
ha_elect_ctx_release_for_test(prev);
ha_fini(&ha);
printf("test_many_prevote_rounds_no_carrier_leak: PASS (20 rounds x 2 votes)\n");
}
/* ── Test (task 3.3): carrier survives supersession while an endpoint is
* still outstanding, freed only once that endpoint's teardown completes ── */
static void test_carrier_survives_supersession_with_outstanding_endpoint(void)
{
ha_state_t ha;
ha_init(&ha, &g_cfg, 0);
ha_elect_ctx_t *round1 = make_test_elect_ctx(2);
/* One vote endpoint captures round1 and is NOT torn down yet. */
endpoint_t *ep = endpoint_get();
TEST_ASSERT_NOT_NULL(ep);
vote_ep_ctx_build_for_test(ep, &ha, round1, 0, 1, 0, 0, "ctrl-a");
atomic_fetch_add_explicit(&round1->refcount, 1, memory_order_relaxed);
vote_ep_wire_on_fini_for_test(ep);
TEST_ASSERT_EQUAL_INT(2, atomic_load_explicit(&round1->refcount, memory_order_relaxed));
/* A new round supersedes round1 (mirrors ha_start_election_async's
* exchange+release at task 2.4) — releases the BIAS reference only. */
ha_elect_ctx_t *round2 = make_test_elect_ctx(2);
ha_elect_ctx_release_for_test(round1); /* bias release: 2 -> 1 */
/* round1 must still be valid — one endpoint reference remains outstanding.
* (We do not read round1's fields here beyond the refcount we already
* captured before this point, matching this suite's convention of never
* dereferencing a carrier that could be at its last reference — but the
* key proof is that NOTHING crashes/UAFs between here and the endpoint's
* teardown below, verified by ASAN.) */
/* Now the outstanding endpoint tears down for real — this must be the
* TRUE last reference and free round1 cleanly. */
endpoint_fini(ep);
free(ep);
/* Clean up round2 (nothing superseded it). */
ha_elect_ctx_release_for_test(round2);
ha_fini(&ha);
printf("test_carrier_survives_supersession_with_outstanding_endpoint: PASS\n");
}
/* ── Test suite entry point ───────────────────────────────────────── */
void test_ha_elect_ctx_lifecycle(void)
{
printf("\n>> ha_elect_ctx_t reference-counted lifecycle tests\n\n");
RUN_TEST(test_release_at_refcount_one_frees_no_crash);
RUN_TEST(test_release_at_refcount_two_leaves_valid);
RUN_TEST(test_many_election_rounds_no_carrier_leak);
RUN_TEST(test_many_prevote_rounds_no_carrier_leak);
RUN_TEST(test_carrier_survives_supersession_with_outstanding_endpoint);
}

8
src/include/netio.h
View file

@ -101,6 +101,13 @@ typedef struct endpoint_t {
time_t ts_last_in;
time_t ts_last_out;
void (*on_connected)(struct endpoint_t *ep); /* fire-once; called in iomgr thread after ssl_handshake_done=1 */
/* ha-elect-ctx-lifecycle task 1.1: fire-once completion-side counterpart
* to on_connected — called by endpoint_fini() immediately before it frees
* ep->data, so a consumer can observe teardown completion (e.g. release a
* reference a struct pointed to by ep->data was holding) while ep->data is
* still valid. Does not change endpoint_fini's teardown ordering/timing —
* it only adds an observation point at the existing free. */
void (*on_fini)(struct endpoint_t *ep);
/* bug-1298: reference count governing the lifetime of a HEAP-allocated
* endpoint (endpoint_get). An ep is fini'd + freed exactly when refcnt
* drops to 0. Holders: the creator/owner (endpoint_get sets refcnt=1, the
@ -176,6 +183,7 @@ extern int endpoint_set_owner(endpoint_t*, void*);
extern int endpoint_set_msg_handler(endpoint_t*, msg_handler_t);
extern int endpoint_set_err_handler(endpoint_t*, err_handler_t);
extern int endpoint_set_on_connected(endpoint_t*, void (*cb)(endpoint_t *));
extern int endpoint_set_on_fini(endpoint_t*, void (*cb)(endpoint_t *));
extern void endpoint_fini(endpoint_t*);
/* bug-1298: endpoint refcount. endpoint_ref takes a reference; endpoint_unref
* drops one and, when the last reference is released, fini's the ep and (for

1
src/lib/meson.build
View file

@ -173,6 +173,7 @@ libxce_test_src = ['testsuite/test.c',
'testsuite/test_cluster_parse.c',
'testsuite/test_libxce_retry.c',
'testsuite/test_netio_on_connected.c',
'testsuite/test_netio_on_fini.c',
'testsuite/test_netio_arm_failure.c',
'testsuite/test_netio_close_propagation.c',
'testsuite/test_netio_bounded_connect.c',

22
src/lib/netio.c
View file

@ -955,6 +955,19 @@ int endpoint_set_on_connected(endpoint_t* ep, void (*cb)(endpoint_t*))
return 0;
}
int endpoint_set_on_fini(endpoint_t* ep, void (*cb)(endpoint_t*))
{
if (ep == NULL) {
return -EINVAL;
}
mylock_lock(&(ep->lock));
ep->on_fini = cb;
mylock_unlock(&(ep->lock));
return 0;
}
int endpoint_enq(endpoint_t* ep, data_item_t* it)
{
int rc = 0;
@ -1091,6 +1104,15 @@ int endpoint_is_tearing_down(endpoint_t* ep)
void endpoint_fini(endpoint_t* ep)
{
dataq_fini(&(ep->outq));
/* ha-elect-ctx-lifecycle task 1.1: fire the completion hook (if set)
* immediately before ep->data is freed, so a consumer can observe
* teardown completion while ep->data is still valid. Does not change
* teardown ordering/timing — same free happens right after, unconditionally. */
if (ep->on_fini) {
ep->on_fini(ep);
}
FREEUP(ep->data);
if (ep->fd >= 0) {

2
src/lib/testsuite/test.c
View file

@ -36,6 +36,7 @@ extern void test_cluster(void);
extern void test_cluster_parse(void);
extern void test_libxce_retry(void);
extern void test_netio_on_connected(void);
extern void test_netio_on_fini(void);
extern void test_netio_arm_failure(void);
extern void test_netio_close_propagation(void);
extern void test_netio_bounded_connect(void);
@ -149,6 +150,7 @@ int main(int argc, char* argv[])
TEST(test_cluster_parse);
TEST(test_libxce_retry);
TEST(test_netio_on_connected);
TEST(test_netio_on_fini);
TEST(test_netio_arm_failure);
TEST(test_netio_close_propagation);
TEST(test_netio_bounded_connect);

169
src/lib/testsuite/test_netio_on_fini.c Normal file
View file

@ -0,0 +1,169 @@
/*
* test_netio_on_fini.c — TDD tests for on_fini completion hook in endpoint_t.
*
* Task 1.1: ha-elect-ctx-lifecycle spec
* Boundary: netio.c / netio.h
*
* Mirrors test_netio_on_connected.c's structure for the completion-side
* counterpart hook. Tests verify:
* 1. on_fini field exists in endpoint_t (compile-time check), NULL by default
* 2. endpoint_set_on_fini() stores the callback
* 3. Hook fires exactly once from endpoint_fini(), before ep->data is freed
* 4. ep->data is freed (NULL'd) after endpoint_fini() returns, as before
* 5. No callback set — endpoint_fini() still frees ep->data, no crash
*/
#define _GNU_SOURCE
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#include "netio.h"
/* ── Static state for callback tracking ──────────────────────────── */
static int g_fini_cb_count = 0;
static endpoint_t *g_fini_cb_ep = NULL;
static void *g_fini_cb_data_seen = NULL;
static void on_fini_stub(endpoint_t *ep)
{
g_fini_cb_count++;
g_fini_cb_ep = ep;
/* Capture ep->data as observed INSIDE the hook — must equal the
* original allocation, proving it is still valid (not yet freed) at
* the moment the hook runs. */
g_fini_cb_data_seen = ep->data;
}
static void reset_cb_state(void)
{
g_fini_cb_count = 0;
g_fini_cb_ep = NULL;
g_fini_cb_data_seen = NULL;
}
/* ── Test 1: Field exists and is zero-initialised after endpoint_get ── */
static void test_on_fini_field_exists_and_is_null(void)
{
endpoint_t *ep = endpoint_get();
TEST_ASSERT_NOT_NULL(ep);
TEST_ASSERT_NULL(ep->on_fini);
free(ep);
printf("test_on_fini_field_exists_and_is_null: PASS\n");
}
/* ── Test 2: endpoint_set_on_fini stores the callback ────────────── */
static void test_endpoint_set_on_fini_stores_callback(void)
{
endpoint_t *ep = endpoint_get();
TEST_ASSERT_NOT_NULL(ep);
int rc = endpoint_set_on_fini(ep, on_fini_stub);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_PTR(on_fini_stub, ep->on_fini);
free(ep);
printf("test_endpoint_set_on_fini_stores_callback: PASS\n");
}
/* ── Test 3: endpoint_set_on_fini with NULL ep returns -EINVAL ───── */
static void test_endpoint_set_on_fini_null_ep(void)
{
int rc = endpoint_set_on_fini(NULL, on_fini_stub);
TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
printf("test_endpoint_set_on_fini_null_ep: PASS\n");
}
/* ── Test 4: endpoint_set_on_fini accepts NULL callback ──────────── */
static void test_endpoint_set_on_fini_null_cb(void)
{
endpoint_t *ep = endpoint_get();
TEST_ASSERT_NOT_NULL(ep);
int rc = endpoint_set_on_fini(ep, NULL);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_NULL(ep->on_fini);
free(ep);
printf("test_endpoint_set_on_fini_null_cb: PASS\n");
}
/*
* ── Test 5: hook fires exactly once from endpoint_fini(), and ep->data
* is still valid (not yet freed) at the moment the hook runs ──────
*/
static void test_on_fini_fires_once_before_data_freed(void)
{
reset_cb_state();
endpoint_t *ep = endpoint_get();
TEST_ASSERT_NOT_NULL(ep);
void *marker = malloc(16);
TEST_ASSERT_NOT_NULL(marker);
ep->data = marker;
endpoint_set_on_fini(ep, on_fini_stub);
TEST_ASSERT_EQUAL_PTR(on_fini_stub, ep->on_fini);
endpoint_fini(ep);
/* Hook fired exactly once, with the correct endpoint. */
TEST_ASSERT_EQUAL_INT(1, g_fini_cb_count);
TEST_ASSERT_EQUAL_PTR(ep, g_fini_cb_ep);
/* ep->data was still the original allocation AT HOOK-CALL TIME —
* proves the hook observes teardown completion before the generic
* FREEUP(ep->data) reclaims it. */
TEST_ASSERT_EQUAL_PTR(marker, g_fini_cb_data_seen);
/* After endpoint_fini() returns, ep->data is freed and NULL'd —
* existing FREEUP(ep->data) behavior is unchanged. */
TEST_ASSERT_NULL(ep->data);
free(ep);
printf("test_on_fini_fires_once_before_data_freed: PASS\n");
}
/* ── Test 6: No callback set — endpoint_fini still frees data, no crash ── */
static void test_no_on_fini_callback_data_still_freed_no_crash(void)
{
endpoint_t *ep = endpoint_get();
TEST_ASSERT_NOT_NULL(ep);
void *marker = malloc(16);
TEST_ASSERT_NOT_NULL(marker);
ep->data = marker;
/* No on_fini set. */
endpoint_fini(ep);
TEST_ASSERT_NULL(ep->data);
free(ep);
printf("test_no_on_fini_callback_data_still_freed_no_crash: PASS\n");
}
/* ── Test suite entry point ───────────────────────────────────────── */
void test_netio_on_fini(void)
{
printf("\n>> netio on_fini completion hook tests\n\n");
RUN_TEST(test_on_fini_field_exists_and_is_null);
RUN_TEST(test_endpoint_set_on_fini_stores_callback);
RUN_TEST(test_endpoint_set_on_fini_null_ep);
RUN_TEST(test_endpoint_set_on_fini_null_cb);
RUN_TEST(test_on_fini_fires_once_before_data_freed);
RUN_TEST(test_no_on_fini_callback_data_still_freed_no_crash);
}