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

fix bug-1060 bug-1067

Browse source
This commit is contained in:
Rongsong Shen 2026-06-24 17:05:43 +08:00
parent 6067ba6c94
commit 61e1391cd0
28 changed files with 7021 additions and 2926 deletions
Download patch file Download diff file Expand all files Collapse all files

80
src/lib/client.c
View file

@ -123,30 +123,59 @@ int xcecli_try_wakeup(xcecli_ctx_t c, int status)
int xcecli_start(xcecli_ctx_t c)
{
int rc;
int rc = -ECONNREFUSED;
api_ctx_t* ctx = (api_ctx_t*)c;
int i;
int connected_idx = -1;
if (c == NULL) {
return -EINVAL;
}
rc = endpoint_connect(&(ctx->ep), ctx->conf->masters[0],
ctx->conf->master_ports[0]);
if (rc < 0) {
/* Initial connect rotates through ALL configured masters, not just
* masters[0] (bug-854 dead-first-master): if the first master is down
* (connection-refused after a failover that killed it), a hardcoded
* masters[0] connect fails the whole `client` session before the failover
* loop — which only runs on an ALREADY-connected ctx — ever gets a chance
* to rotate. Try each master in order; the first that connects becomes the
* sticky cursor start. The subsequent ctrlstatus/op failover loop then
* rotates onward to the authoritative primary as needed.
*
* ctx->ep was endpoint_init'd + bound to the iomgr in xcecli_init. A failed
* endpoint_connect leaves the socket fd unusable, so before retrying the
* NEXT master we must rebuild the endpoint exactly as xce_ctx_reconnect
* does: endpoint_fini → endpoint_init (fresh fd) → re-set peer + re-bind the
* iomgr (endpoint_connect needs ep->iomgr->ssl_ctx). */
for (i = 0; i < ctx->conf->nmasters; i++) {
if (i > 0) {
endpoint_fini(&(ctx->ep));
if (endpoint_init(&(ctx->ep), AF_INET, SOCK_STREAM, 0, 0) < 0)
continue;
endpoint_set_peer(&(ctx->ep), &(ctx->ctx.myinfo));
ctx->ep.iomgr = &(ctx->ctx.iomgr);
}
rc = endpoint_connect(&(ctx->ep), ctx->conf->masters[i],
ctx->conf->master_ports[i]);
if (rc >= 0) {
connected_idx = i;
break;
}
}
if (connected_idx < 0) {
/* No master reachable at all — surface the last transport error. */
endpoint_fini(&(ctx->ep));
return rc;
}
/* Initialize per-context failover state: sticky primary starts at
* masters[0] (the initial connect target above), retry tunables default
* to XCE_RETRY_DEFAULTS, and no rejoin hook is installed for pure clients.
*/
/* Initialize per-context failover state: sticky primary starts at the master
* we actually connected to (not always index 0), retry tunables default to
* XCE_RETRY_DEFAULTS, and no rejoin hook is installed for pure clients. */
rc = mylock_init(&(ctx->failover.lock));
if (rc < 0) {
endpoint_fini(&(ctx->ep));
return rc;
}
ctx->failover.cur_master = 0;
ctx->failover.cur_master = connected_idx;
ctx->failover.retry = XCE_RETRY_DEFAULTS;
ctx->failover.rejoin_cb = NULL;
ctx->failover.rejoin_arg = NULL;
@ -414,11 +443,17 @@ int xcecli_ctrlstatus_primary(xcecli_ctx_t c, xce_ctrlstatus_t** status)
/* Bound the redirect loop by the configured failover retry budget so we
* never spin unbounded chasing a never-appearing primary (each inner
* xcecli_ctrlstatus already runs its own full reach-some-node failover). */
* xcecli_ctrlstatus already runs its own full reach-some-node failover).
* Floor it at the master-ring size so a standby-first / dead-first master
* list is always traversable in full before giving up (bug-854): one extra
* attempt lets the rotation that lands on the primary actually query it. */
max_attempts = ctx->failover.retry.max_retries;
if (max_attempts < 1) {
max_attempts = 1;
}
if (ctx->conf != NULL && ctx->conf->nmasters + 1 > max_attempts) {
max_attempts = ctx->conf->nmasters + 1;
}
for (attempt = 0; attempt < max_attempts; attempt++) {
hint = NULL;
@ -436,21 +471,34 @@ int xcecli_ctrlstatus_primary(xcecli_ctx_t c, xce_ctrlstatus_t** status)
return 0;
}
/* Non-authoritative HINT (standby/electing/init). Direct the next query
* at the reported primary if we can map it to an address; otherwise let
* the next xcecli_ctrlstatus rotate the configured master list. The
* redirect hint is consumed one-shot by xce_failover_next_target. */
/* Non-authoritative reply (standby/electing/init). The inner
* xcecli_ctrlstatus SUCCEEDS against ANY reachable node — ctrlstatus is
* answerable in every role — so it classifies the standby's status-0
* reply as success and NEVER rotates off it on its own. We must advance
* the endpoint here, or the loop re-queries the same non-primary every
* attempt and exhausts the budget without ever reaching the primary
* (bug-854). Capture a redirect hint when the responder named the primary
* (1-hop direct redirect), then FORCE rotation: xce_failover_rotate
* consumes the hint when set, else advances to the next configured
* master; recover repairs the ctx so the next attempt's send proceeds.
* Mirror xcecli_retry_next: only recover when the rotate succeeded — a
* failed rotate leaves the ctx for the next inner call's own transport-
* error failover to heal. All under failover.lock (the inner call takes
* and releases it itself, so it is not held here). */
{
char phost[XCE_MAXHOSTNAME_LEN];
uint16_t pport = 0;
mylock_lock(&ctx->failover.lock);
if (xcecli_ctrlstatus_primary_addr(hint, phost, sizeof(phost),
&pport)
== 0) {
mylock_lock(&ctx->failover.lock);
xce_ctx_capture_redirect_hint(&ctx->ctx, phost, pport);
mylock_unlock(&ctx->failover.lock);
}
if (xce_failover_rotate(ctx) == 0) {
xcecli_ctx_recover(ctx);
}
mylock_unlock(&ctx->failover.lock);
}
/* Free the non-authoritative interim reply before the next attempt — no

27
src/lib/hook.c
View file

@ -42,20 +42,35 @@ int xce_add_hook(hooksys_t* s, int typ, hook_fn_t fn, void* data)
return -EINVAL;
}
/* Scan for a disabled (NULL-fn) slot of the same type to reuse */
/* Single scan of the type's slots:
* - if (fn,data) is ALREADY registered and active, adding again is a
* no-op (idempotent). xce_disable_hook removes only ONE matching slot
* and xce_run_hooks fires ALL slots, so a duplicate (fn,data) would
* fire twice and survive a single disable — the bug-845 double-append
* (persist_start + ha_register_persist_hooks both register the same
* persist_job_hook_fun closure). Dedup is keyed on (fn,data): the same
* fn with a DISTINCT closure stays a legitimate second registration.
* - otherwise remember the first disabled (NULL-fn) slot to reuse. */
{
xlist_t* l = &s->hooks[typ];
int i;
xlist_t* l = &s->hooks[typ];
hook_item_t* reuse = NULL;
int i;
for (i = 0; i < l->count; i++) {
hook_item_t* slot = (hook_item_t*)l->data[i];
if (!slot) continue;
hook_fn_t cur =
atomic_load_explicit(&slot->fn, memory_order_acquire);
if (cur == NULL) {
slot->data = data;
atomic_store_explicit(&slot->fn, fn, memory_order_release);
return 0;
if (!reuse) reuse = slot;
continue;
}
if (cur == fn && slot->data == data)
return 0; /* already registered — no duplicate */
}
if (reuse != NULL) {
reuse->data = data;
atomic_store_explicit(&reuse->fn, fn, memory_order_release);
return 0;
}
}

8
src/lib/request.c
View file

@ -486,6 +486,11 @@ static void wal_entry_to_wire(struct capn_segment *cs,
w->prevLogIndex = e->prev_log_index;
w->prevLogTerm = e->prev_log_term;
w->leaderCommit = e->leader_commit;
/* bug-1067: carry the streaming leader's CURRENT term per sub-entry (distinct
* from `term` above, the entry's stored term). The standby intake uses this as
* the AppendEntries header term so historical (entry_term < leader_term)
* catch-up entries are not stale-rejected. */
w->leaderCurrentTerm = e->leader_current_term;
if (e->raw_payload && e->raw_payload_len > 0) {
capn_list8 pl = capn_new_list8(cs, (int)e->raw_payload_len);
capn_setv8(pl, 0, e->raw_payload, (int)e->raw_payload_len);
@ -539,6 +544,9 @@ int xce_wal_batch_encode_raw(wal_catchup_entry_t **entries, int n_entries,
top.prevLogIndex = entries[0]->prev_log_index;
top.prevLogTerm = entries[0]->prev_log_term;
top.leaderCommit = entries[0]->leader_commit;
/* bug-1067: the AppendEntries header term (leader's CURRENT term), distinct
* from top.term (entries[0]'s stored term). Mirrors the per-entry field. */
top.leaderCurrentTerm = entries[0]->leader_current_term;
/* Batch container: entries list carries the per-entry payload(s) (empty for
* a heartbeat). Each sub-entry carries lsn/entryType/term/payload. */

55
src/lib/testsuite/test_hooks.c
View file

@ -100,6 +100,59 @@ static void test_add_hook_reuses_null_slot(void)
printf("test_add_hook_reuses_null_slot: PASS\n");
}
/* bug-845: re-registering the SAME (fn,data) on a type must be idempotent.
* xce_disable_hook removes only ONE matching slot and xce_run_hooks fires ALL
* slots, so a duplicate (fn,data) registration would (a) fire twice and (b)
* survive a single disable. This is exactly how the controller double-appended
* every WAL_JOB_SUBMIT: persist_start() and ha_register_persist_hooks() both
* register persist_job_hook_fun with the same persist_t closure. */
static void test_add_hook_dedups_same_fn_and_data(void)
{
hooksys_t hs;
int a = 0;
TEST_ASSERT_EQUAL_INT(0, xce_hook_init(&hs, 2));
/* First add registers; the second add with the SAME fn AND data must be a
* no-op (still rc=0, still exactly one slot). */
TEST_ASSERT_EQUAL_INT(0, xce_add_hook(&hs, HOOK_AFTER_EVENT, counting_fn, &a));
TEST_ASSERT_EQUAL_INT(0, xce_add_hook(&hs, HOOK_AFTER_EVENT, counting_fn, &a));
TEST_ASSERT_EQUAL_INT(1, hs.hooks[HOOK_AFTER_EVENT].count);
/* Fires ONCE per event, never twice. */
xce_run_hooks(&hs, HOOK_AFTER_EVENT, 0, 0);
TEST_ASSERT_EQUAL_INT(1, a);
/* A single disable fully removes it — no duplicate slot lingers. */
TEST_ASSERT_EQUAL_INT(0, xce_disable_hook(&hs, HOOK_AFTER_EVENT, counting_fn));
xce_run_hooks(&hs, HOOK_AFTER_EVENT, 0, 0);
TEST_ASSERT_EQUAL_INT(1, a); /* unchanged */
xce_hook_fini(&hs);
printf("test_add_hook_dedups_same_fn_and_data: PASS\n");
}
/* The dedup is keyed on (fn,data): the SAME fn with DISTINCT closures is a
* legitimate pair of registrations and BOTH must fire. */
static void test_add_hook_allows_same_fn_distinct_data(void)
{
hooksys_t hs;
int a = 0, b = 0;
TEST_ASSERT_EQUAL_INT(0, xce_hook_init(&hs, 2));
TEST_ASSERT_EQUAL_INT(0, xce_add_hook(&hs, HOOK_AFTER_EVENT, counting_fn, &a));
TEST_ASSERT_EQUAL_INT(0, xce_add_hook(&hs, HOOK_AFTER_EVENT, counting_fn, &b));
TEST_ASSERT_EQUAL_INT(2, hs.hooks[HOOK_AFTER_EVENT].count);
xce_run_hooks(&hs, HOOK_AFTER_EVENT, 0, 0);
TEST_ASSERT_EQUAL_INT(1, a);
TEST_ASSERT_EQUAL_INT(1, b);
xce_hook_fini(&hs);
printf("test_add_hook_allows_same_fn_distinct_data: PASS\n");
}
/* --- suite entry point --- */
void test_hooks(void)
@ -109,4 +162,6 @@ void test_hooks(void)
RUN_TEST(test_disable_hook_stops_invocation);
RUN_TEST(test_disable_hook_returns_enoent_if_not_found);
RUN_TEST(test_add_hook_reuses_null_slot);
RUN_TEST(test_add_hook_dedups_same_fn_and_data);
RUN_TEST(test_add_hook_allows_same_fn_distinct_data);
}

5
src/lib/testsuite/test_wal_wire.c
View file

@ -191,8 +191,11 @@ static void test_wal_batch_encode_rejects_invalid_args(void)
TEST_ASSERT_EQUAL_INT(-EINVAL,
xce_wal_batch_encode_raw(NULL, 1, &test_peer, 0, &pkt));
/* n_entries == 0 is a valid heartbeat (ha-term-safety task 3.2): a 0-entry AE
* carries the AppendEntries metadata (term, prev tuple, leader_commit) on the
* top-level container to drive UpdateTerm/logOk. Only n_entries < 0 is invalid. */
TEST_ASSERT_EQUAL_INT(-EINVAL,
xce_wal_batch_encode_raw(entries, 0, &test_peer, 0, &pkt));
xce_wal_batch_encode_raw(entries, -1, &test_peer, 0, &pkt));
TEST_ASSERT_EQUAL_INT(-EINVAL,
xce_wal_batch_encode_raw(entries, 1, &test_peer, 0, NULL));

12
src/lib/xcectx.c
View file

@ -7,6 +7,7 @@
#include "netio.h"
#include "typemap.h"
#include "utils.h"
#include "controller.capnp.h" /* free_CtrlStatus_ptr, free_ClusterConf_ptr */
#include "xcectx.h"
static int handle_get_reply(xce_ctx_t* ctx, xce_reply_t* reply)
@ -71,6 +72,14 @@ static int handle_reply(xce_ctx_t* ctx, xce_reply_t* reply)
case OPCode_clusterconf:
rc = -1;
if (reply->data_which == RequestReply_data_clusterconf) {
/* Free a prior unconsumed reply before storing the new one.
* The consumer xce_clusterconf() MOVE_PTRs out, so in the normal
* single-request-per-ctx path this is a no-op; on a re-stolen slot
* (reply decode on a stale fd after client rotation) it prevents
* a permanent heap leak of the prior occupant. */
if (ctx->reply.u.clusterconf) {
free_ClusterConf_ptr(&ctx->reply.u.clusterconf);
}
MOVE_PTR(ctx->reply.u.clusterconf, reply->data.clusterconf);
rc = 1;
}
@ -78,6 +87,9 @@ static int handle_reply(xce_ctx_t* ctx, xce_reply_t* reply)
case OPCode_ctrlstatus:
rc = -1;
if (reply->data_which == RequestReply_data_ctrlstatus) {
if (ctx->reply.u.ctrlstatus) {
free_CtrlStatus_ptr(&ctx->reply.u.ctrlstatus);
}
MOVE_PTR(ctx->reply.u.ctrlstatus, reply->data.ctrlstatus);
rc = 1;
}