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

feat(ha-repl-hardening): wire config push from decision loop (task 3.3)

Browse source 
ha_dispatch_repl_action_for_standby() calls push_config_to_standbys()
when PUSH_CONFIG decision. Hook seam for unit tests. Wired into
repl_ack_decode_and_dispatch after ha_process_standby_state.
oldest_wal_lsn=0 placeholder documented for task 3.4.
This commit is contained in:
Rongsong Shen 2026-06-04 17:23:46 +00:00
parent a475ef621f
commit d1c7872d73
4 changed files with 258 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.kiro/specs/ha-repl-hardening/tasks.md
View file

@ -90,7 +90,7 @@
- _Boundary: Primary Heartbeat Handler (ha.c)_
- _Depends: 3.1_
- [ ] 3.3 Trigger per-standby config push from decision loop
- [x] 3.3 Trigger per-standby config push from decision loop
- When `standby_tag < committed_tag`, call `push_config_to_standbys()` targeting the specific standby
- Reuse existing config push path; trigger is now heartbeat decision loop instead of `handle_ctrlhello`
- After config push, primary skips further decisions for this standby until the next heartbeat reply confirms the updated tag

106
src/controller/ha.c
View file

@ -1009,6 +1009,92 @@ void ha_process_standby_state(ha_state_t *ha, int peer_idx,
}
}
/*
* ha_dispatch_repl_action_for_standby — execute the decision-loop action for
* one standby on the current heartbeat cycle (Task 3.3).
*
* Reads ha->standby_reported_tag[peer_idx] / standby_reported_lsn[peer_idx]
* (stored by ha_process_standby_state on the same heartbeat reply), calls
* ha_evaluate_repl_action, and acts on the result:
*
* HA_REPL_PUSH_CONFIG → push_config_to_standbys(c, …) [or test hook]
* HA_REPL_SEND_SNAPSHOT→ stub (Task 3.4)
* HA_REPL_STREAM_WAL → stub (Task 3.5)
* HA_REPL_CONVERGED → no-op
*
* One action per heartbeat cycle per standby (design §Unified Heartbeat-Driven
* Protocol). Returns 0 on success, -EINVAL on bad peer_idx.
*
* Requirements: 1.4 (ha-repl-hardening spec)
*/
int ha_dispatch_repl_action_for_standby(ha_state_t *ha,
cluster_t *c,
int peer_idx,
uint64_t committed_tag,
uint64_t committed_lsn,
uint64_t oldest_wal_lsn)
{
if (!ha || peer_idx < 0 || peer_idx >= MAX_HA_CANDIDATES)
return -EINVAL;
uint64_t standby_tag = ha->standby_reported_tag[peer_idx];
uint64_t standby_lsn = ha->standby_reported_lsn[peer_idx];
ha_repl_action_t action = ha_evaluate_repl_action(standby_tag, standby_lsn,
committed_tag, committed_lsn,
oldest_wal_lsn);
switch (action) {
case HA_REPL_PUSH_CONFIG:
INFO("HA repl: standby[%d] tag=%" PRIu64 " < committed_tag=%" PRIu64
" → push config", peer_idx, standby_tag, committed_tag);
#ifdef XCE_TESTING
if (ha_repl_push_config_for_standby_hook != NULL) {
ha_repl_push_config_for_standby_hook(c, peer_idx);
break;
}
#endif
if (c) {
#ifndef XCE_TESTING
extern int conf_read_raw(const char *, uint8_t **, size_t *);
uint8_t *cfg_bytes = NULL;
size_t cfg_len = 0;
if (c->myconf && c->myconf->conffile &&
conf_read_raw(c->myconf->conffile, &cfg_bytes, &cfg_len) == 0) {
push_config_to_standbys(c, cfg_bytes, cfg_len,
c->myconf->config_sha256,
c->myconf->config_version_tag);
free(cfg_bytes);
} else {
push_config_to_standbys(c, NULL, 0, NULL,
c->myconf ? c->myconf->config_version_tag : 0);
}
#endif
}
break;
case HA_REPL_SEND_SNAPSHOT:
/* Task 3.4: snapshot transfer — stub */
INFO("HA repl: standby[%d] lsn=%" PRIu64
" < oldest_wal_lsn=%" PRIu64 " → snapshot (stub)",
peer_idx, standby_lsn, oldest_wal_lsn);
break;
case HA_REPL_STREAM_WAL:
/* Task 3.5: WAL streaming — stub */
INFO("HA repl: standby[%d] lsn=%" PRIu64
" < committed_lsn=%" PRIu64 " → stream WAL (stub)",
peer_idx, standby_lsn, committed_lsn);
break;
case HA_REPL_CONVERGED:
/* No action needed — standby is up to date */
break;
}
return 0;
}
/* ── Dispatch gate (req 1.3, 1.4) ──────────────────────────────────────────
*
* Called from myreq_handler() in cluster.c before the opcode switch.
@ -1537,6 +1623,15 @@ int (*ha_push_config_hook)(void *c) = NULL;
*/
int (*ha_nop_req_hook)(endpoint_t *ep) = NULL;
/*
* ha_repl_push_config_for_standby_hook: test-only seam for the per-standby
* config push triggered by the heartbeat-reply decision loop (Task 3.3).
* When non-NULL, ha_dispatch_repl_action_for_standby() calls this INSTEAD OF
* push_config_to_standbys() when the decision is HA_REPL_PUSH_CONFIG.
* Requirements: 1.4 (ha-repl-hardening)
*/
int (*ha_repl_push_config_for_standby_hook)(void *c, int peer_idx) = NULL;
#endif /* XCE_TESTING */
/* ── Election Evaluation (Req 2.1, 2.7, 2.8) ───────────────────────────────
@ -2114,6 +2209,17 @@ static int repl_ack_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx)
ha_process_standby_state(ha, peer_idx,
sr.config_tag, sr.applied_lsn,
committed_tag, committed_lsn);
/* Task 3.3: run the unified decision loop for this standby.
* oldest_wal_lsn=0 until snapshot transfer is wired (Task 3.4). */
uint64_t oldest_wal_lsn = 0;
#ifndef XCE_TESTING
if (c)
oldest_wal_lsn = 0; /* Task 3.4: wire persist->oldest_wal_lsn */
#endif
ha_dispatch_repl_action_for_standby(ha, c, peer_idx,
committed_tag, committed_lsn,
oldest_wal_lsn);
} else {
capn_free(&capn_ctx);
}

45
src/controller/ha.h
View file

@ -529,6 +529,36 @@ void ha_start_election_async(ha_state_t *ha, cluster_t *c,
*/
void ha_repl_ack_received(ha_state_t *ha, cluster_t *c);
/*
* ha_dispatch_repl_action_for_standby: evaluate the replication decision for
* one standby and execute the resulting action.
*
* Reads ha->standby_reported_tag[peer_idx] and ha->standby_reported_lsn[peer_idx]
* (stored by ha_process_standby_state on the previous call), calls
* ha_evaluate_repl_action, and if the result is HA_REPL_PUSH_CONFIG:
* - In production builds: calls push_config_to_standbys(c, …).
* - In XCE_TESTING builds: calls ha_repl_push_config_for_standby_hook(c, peer_idx)
* if the hook is non-NULL.
*
* One action per heartbeat cycle per standby (design §Unified Heartbeat-Driven
* Protocol). Other actions (SEND_SNAPSHOT, STREAM_WAL) are stubs returning 0
* pending Tasks 3.4–3.5; CONVERGED is a no-op.
*
* peer_idx — 0-based index into ha->standby_reported_* arrays
* committed_tag — primary's committed config_tag (from c->myconf or test fixture)
* committed_lsn — primary's committed WAL LSN (from c->persist or test fixture)
* oldest_wal_lsn— oldest retained WAL LSN (0 = all WAL retained, skip snapshot)
*
* Returns 0 on success; negative on internal error.
* Requirements: 1.4 (ha-repl-hardening spec)
*/
int ha_dispatch_repl_action_for_standby(ha_state_t *ha,
cluster_t *c,
int peer_idx,
uint64_t committed_tag,
uint64_t committed_lsn,
uint64_t oldest_wal_lsn);
/*
* ha_send_ctrlhello_to_all_standbys: iterate ha->standby_endpoints and send
* an OP_hello CtrlHelloRequest to each connected standby.
@ -565,6 +595,21 @@ extern int (*ha_send_ctrlhello_hook)(void *c);
*/
extern int (*ha_push_config_hook)(void *c);
/*
* ha_repl_push_config_for_standby_hook: test-only seam for the per-standby
* config push triggered by the heartbeat-reply decision loop (Task 3.3).
*
* When non-NULL, ha_dispatch_repl_action_for_standby() calls this INSTEAD OF
* push_config_to_standbys() when the decision is HA_REPL_PUSH_CONFIG, so unit
* tests can verify the wiring without requiring a real cluster_t or network.
*
* Signature: int hook(void *c, int peer_idx)
* c — cluster_t* (may be NULL in test mode)
* peer_idx — 0-based index of the standby that triggered the push
* Requirements: 1.4 (ha-repl-hardening)
*/
extern int (*ha_repl_push_config_for_standby_hook)(void *c, int peer_idx);
/*
* ha_tcp_connect_hook: test-only seam for the raw TCP connect inside
* ha_find_and_connect_to_primary. When non-NULL, the function calls this

106
src/controller/testsuite/test_ha_role.c
View file

@ -377,6 +377,109 @@ static void test_process_standby_state_multiple_peers(void)
printf("test_process_standby_state_multiple_peers: PASS\n");
}
/* ── Task 3.3: ha_dispatch_repl_action_for_standby PUSH_CONFIG wiring ──────
*
* ha_dispatch_repl_action_for_standby reads the standby's stored
* (standby_reported_tag, standby_reported_lsn), calls ha_evaluate_repl_action,
* and if the result is HA_REPL_PUSH_CONFIG calls push_config_to_standbys (in
* production) or ha_repl_push_config_for_standby_hook (in XCE_TESTING).
*
* Requirements: 1.4 (ha-repl-hardening), design §Unified Heartbeat-Driven Protocol
*/
#ifdef XCE_TESTING
extern int (*ha_repl_push_config_for_standby_hook)(void *c, int peer_idx);
#endif
/* Call-counting state for the hook */
static int g_push_config_for_standby_count = 0;
static int g_push_config_for_standby_idx = -1;
static int stub_repl_push_config_for_standby(void *c, int peer_idx)
{
(void)c;
g_push_config_for_standby_count++;
g_push_config_for_standby_idx = peer_idx;
return 0;
}
/*
* Test: when standby_reported_tag < committed_tag, ha_dispatch_repl_action_for_standby
* must call push_config_to_standbys (via hook in test mode).
*
* Steps:
* 1. Set ha->standby_reported_tag[0] = 3 (behind committed_tag=5)
* 2. Set ha->standby_reported_lsn[0] = 100 (at committed_lsn)
* 3. Call ha_dispatch_repl_action_for_standby(ha, NULL, 0, committed_tag=5, ...)
* 4. Assert hook called once with peer_idx=0
*
* Requirements: 1.4 (push_config when standby tag < committed_tag)
*/
static void test_dispatch_repl_action_push_config_when_tag_behind(void)
{
#ifdef XCE_TESTING
ha_repl_push_config_for_standby_hook = stub_repl_push_config_for_standby;
g_push_config_for_standby_count = 0;
g_push_config_for_standby_idx = -1;
ha_state_t *ha = build_ha_for_convergence_test();
/* standby[0] has stale config tag */
ha->standby_reported_tag[0] = 3;
ha->standby_reported_lsn[0] = 100;
int rc = ha_dispatch_repl_action_for_standby(
ha, /*c=*/NULL, /*peer_idx=*/0,
/*committed_tag=*/5, /*committed_lsn=*/100, /*oldest_wal_lsn=*/0);
(void)rc;
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_push_config_for_standby_count,
"PUSH_CONFIG: hook must fire once when standby_tag < committed_tag");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_push_config_for_standby_idx,
"PUSH_CONFIG: hook must receive correct peer_idx=0");
ha_fini(ha);
ha_repl_push_config_for_standby_hook = NULL;
printf("test_dispatch_repl_action_push_config_when_tag_behind: PASS\n");
#else
printf("test_dispatch_repl_action_push_config_when_tag_behind: SKIP (non-test build)\n");
#endif
}
/*
* Test: when standby is converged (tag == committed, lsn == committed),
* ha_dispatch_repl_action_for_standby must NOT call push_config hook.
*
* Requirements: decision tree — CONVERGED does not trigger push
*/
static void test_dispatch_repl_action_no_push_when_converged(void)
{
#ifdef XCE_TESTING
ha_repl_push_config_for_standby_hook = stub_repl_push_config_for_standby;
g_push_config_for_standby_count = 0;
g_push_config_for_standby_idx = -1;
ha_state_t *ha = build_ha_for_convergence_test();
/* standby[0] is fully converged */
ha->standby_reported_tag[0] = 5;
ha->standby_reported_lsn[0] = 100;
int rc = ha_dispatch_repl_action_for_standby(
ha, /*c=*/NULL, /*peer_idx=*/0,
/*committed_tag=*/5, /*committed_lsn=*/100, /*oldest_wal_lsn=*/0);
(void)rc;
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_push_config_for_standby_count,
"CONVERGED: hook must NOT fire when standby is already converged");
ha_fini(ha);
ha_repl_push_config_for_standby_hook = NULL;
printf("test_dispatch_repl_action_no_push_when_converged: PASS\n");
#else
printf("test_dispatch_repl_action_no_push_when_converged: SKIP (non-test build)\n");
#endif
}
/* ── Task 3.2: ha_evaluate_repl_action unified decision tree ─────────────── */
/*
@ -491,6 +594,9 @@ void test_ha_role(void)
RUN_TEST(test_evaluate_repl_action_converged);
RUN_TEST(test_evaluate_repl_action_send_snapshot);
RUN_TEST(test_evaluate_repl_action_stream_wal);
/* Task 3.3: ha_dispatch_repl_action_for_standby PUSH_CONFIG wiring */
RUN_TEST(test_dispatch_repl_action_push_config_when_tag_behind);
RUN_TEST(test_dispatch_repl_action_no_push_when_converged);
}
/* test_am_i_primary suite deleted — am_i_primary() removed (election-iomgr-refactor) */