feat(ha-repl-hardening): safe_committed_lsn from vote replies (task 4.1)

- Add ha_compute_safe_committed_lsn() in ha.c: collects LSNs from
  granted voters (self + peers), sorts descending, picks index
  floor(n_granted/2) for majority-threshold value
- Wire into raft_cb_become_leader: compute safe_lsn/safe_tag from
  ha->elect_ctx after winning election, call persist_truncate_wal_above
- Four unit tests in test_ha_evaluate_committing.c:
  voters [99,100]+self=100 → 100; voters [50,51]+self=100 → 51;
  single-node → 0; denied voter excluded → self-only
This commit is contained in:
Rongsong Shen 2026-06-05 02:41:22 +00:00
parent 18ad3e9c9f
commit e7919304c3
4 changed files with 314 additions and 1 deletions

View file

@ -341,6 +341,94 @@ static void raft_cb_start_election_after_delay(void *ctx, uint32_t delay_ms)
}
}
/* ── Task 4.1 (ha-repl-hardening): safe commit point computation ─────────────
*
* ha_compute_safe_committed_lsn: determine the safe committed LSN and config
* tag after winning an election, using the vote replies collected in ctx.
*
* Algorithm (design §Flow 0: Post-Election — Determine Safe Commit Point):
* 1. Collect LSNs from all granted voters:
* - Self always granted: self_lsn = votes[0].my_lsn (when n_peers > 0)
* - Each peer with vote_granted=1: lsn = votes[i].peer_lsn
* 2. Sort granted LSNs descending.
* 3. safe_committed_lsn = sorted[floor(n_granted / 2)]
* (the majority-threshold element — the highest LSN held by a majority)
* 4. Same logic for config tags (votes[i].my_tag / votes[i].peer_tag).
*
* Single-node (n_peers == 0): votes[] is empty, outputs 0 for both fields.
*
* Requirements: 2.3 (ha-repl-hardening spec)
* Boundary: Controller FSM (ha.c)
*/
static int _cmp_u64_desc(const void *a, const void *b)
{
uint64_t ua = *(const uint64_t *)a;
uint64_t ub = *(const uint64_t *)b;
if (ua > ub) return -1;
if (ua < ub) return 1;
return 0;
}
void ha_compute_safe_committed_lsn(ha_elect_ctx_t *ctx,
uint64_t *safe_lsn_out,
uint64_t *safe_tag_out)
{
if (!ctx || !safe_lsn_out || !safe_tag_out) return;
/* Single-node: no votes[] entries — self is the only majority. */
if (ctx->n_peers == 0) {
*safe_lsn_out = 0;
*safe_tag_out = 0;
return;
}
/* Collect granted voter LSNs and tags.
* Maximum granted voters = n_peers + 1 (self + all other peers). */
int max_voters = ctx->n_peers + 1;
uint64_t *lsns = calloc((size_t)max_voters, sizeof(uint64_t));
uint64_t *tags = calloc((size_t)max_voters, sizeof(uint64_t));
if (!lsns || !tags) {
free(lsns);
free(tags);
*safe_lsn_out = 0;
*safe_tag_out = 0;
return;
}
int n_granted = 0;
/* Self always granted: self's lsn/tag lives in votes[0].my_lsn / my_tag */
lsns[n_granted] = ctx->votes[0].my_lsn;
tags[n_granted] = ctx->votes[0].my_tag;
n_granted++;
/* Each peer with vote_granted=1 */
for (int i = 0; i < ctx->n_peers; i++) {
if (ctx->votes[i].vote_granted) {
lsns[n_granted] = ctx->votes[i].peer_lsn;
tags[n_granted] = ctx->votes[i].peer_tag;
n_granted++;
}
}
/* Sort descending */
qsort(lsns, (size_t)n_granted, sizeof(uint64_t), _cmp_u64_desc);
qsort(tags, (size_t)n_granted, sizeof(uint64_t), _cmp_u64_desc);
/* Majority-threshold element: index = floor(n_granted / 2)
* For n_granted=3: index=1 (2nd element, 0-based) — majority=2 out of 3.
* For n_granted=1: index=0 (self only). */
int idx = n_granted / 2;
*safe_lsn_out = lsns[idx];
*safe_tag_out = tags[idx];
free(lsns);
free(tags);
INFO("HA: safe_committed_lsn=%" PRIu64 " safe_config_tag=%" PRIu64
" (from %d granted voters)", *safe_lsn_out, *safe_tag_out, n_granted);
}
/*
* raft_cb_become_leader: node won the election — enter COMMITTING state.
*
@ -363,6 +451,29 @@ static void raft_cb_become_leader(void *ctx, uint64_t term)
/* Sync ha->term from raft term (raft owns term increments) */
atomic_store_explicit(&ha->term, term, memory_order_release);
/* Task 4.1 (ha-repl-hardening): compute safe_committed_lsn from vote replies.
* Uses voter LSNs collected during election (ha->elect_ctx->votes[]) to find
* the highest LSN held by a majority without an extra round-trip.
* Design §Flow 0: Post-Election — Determine Safe Commit Point. */
{
ha_elect_ctx_t *ectx = (ha_elect_ctx_t *)atomic_load_explicit(
&ha->elect_ctx, memory_order_acquire);
if (ectx) {
uint64_t safe_lsn = 0;
uint64_t safe_tag = 0;
ha_compute_safe_committed_lsn(ectx, &safe_lsn, &safe_tag);
/* Store results for use in WAL truncation and COMMITTING phase.
* safe_lsn is the highest LSN on a majority; entries above it
* may not have been committed on all nodes and must be truncated. */
if (c) {
persist_truncate_wal_above(&c->persist, safe_lsn);
}
/* Update committed config tag baseline for the COMMITTING decision loop */
atomic_store_explicit(&ha->committed_config_tag, safe_tag,
memory_order_release);
}
}
int is_first_boot = !ha->phase2_done;
/* First boot: call cluster_init_phase2 */