diff --git a/.kiro/specs/ha-repl-hardening/tasks.md b/.kiro/specs/ha-repl-hardening/tasks.md index ce6314d4..d3107ea3 100644 --- a/.kiro/specs/ha-repl-hardening/tasks.md +++ b/.kiro/specs/ha-repl-hardening/tasks.md @@ -147,7 +147,7 @@ --- - [ ] 4. Post-election safety: Safe commit point from vote replies -- [ ] 4.1 Determine `safe_committed_lsn` from vote replies in `raft_cb_become_leader` +- [x] 4.1 Determine `safe_committed_lsn` from vote replies in `raft_cb_become_leader` - `ha_elect_ctx_t.votes[]` is already populated by the vote reply handler (`peerLastLogLsn`, `peerConfigTag` per voter); this task reads those fields after winning election - Compute `safe_committed_lsn` = max LSN held by a majority of voters (iterate all voters with `granted=true`) - Compute `safe_config_tag` from `peerConfigTag` using the same majority logic diff --git a/src/controller/ha.c b/src/controller/ha.c index c011077a..5a1b4f8c 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -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 */ diff --git a/src/controller/ha.h b/src/controller/ha.h index cc369f63..002495ff 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -289,6 +289,22 @@ int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p, void ha_nop_reply_state_report(ha_state_t *ha, myconf_t *myconf, state_report_t *sr, xce_reply_t *reply_out); +/* ── Task 4.1 (ha-repl-hardening): safe commit point computation ───────────── + * + * ha_compute_safe_committed_lsn: determine safe_committed_lsn and safe_config_tag + * from vote replies collected in ctx after winning an election. + * + * Collects LSNs from all granted voters (self + peers with vote_granted=1), + * sorts descending, and returns the majority-threshold element. + * + * single-node (n_peers=0): outputs 0 for both fields (no vote data available). + * + * Requirements: 2.3 (ha-repl-hardening spec) + */ +void ha_compute_safe_committed_lsn(ha_elect_ctx_t *ctx, + uint64_t *safe_lsn_out, + uint64_t *safe_tag_out); + /* ── Task 3.2: Unified replication decision enum and evaluator ────────────── */ /* diff --git a/src/controller/testsuite/test_ha_evaluate_committing.c b/src/controller/testsuite/test_ha_evaluate_committing.c index 7146d271..daad35e7 100644 --- a/src/controller/testsuite/test_ha_evaluate_committing.c +++ b/src/controller/testsuite/test_ha_evaluate_committing.c @@ -427,6 +427,187 @@ static void test_committing_timeout_fn_cancelled_is_noop(void) printf("test_committing_timeout_fn_cancelled_is_noop: PASS (role still CTRL_PRIMARY_COMMITTING)\n"); } +/* ── Task 4.1 (ha-repl-hardening): ha_compute_safe_committed_lsn ──────────── + * + * Computes safe_committed_lsn and safe_config_tag from the vote replies stored + * in ha_elect_ctx_t.votes[] after winning an election. + * + * Algorithm: collect LSNs from all granted voters (including self at votes[0].my_lsn), + * sort descending, take the value at index floor(n_granted/2) (i.e. the majority + * threshold element) — this is the highest LSN held by a majority. + * + * Same logic applies for config tags (peer_tag). + * + * Requirements: 2.3 (ha-repl-hardening spec) + * Design ref: design.md §Flow 0: Post-Election — Determine Safe Commit Point + */ +#ifdef XCE_TESTING +extern void ha_compute_safe_committed_lsn(ha_elect_ctx_t *ctx, + uint64_t *safe_lsn_out, + uint64_t *safe_tag_out); +#endif + +/* + * Test case A (from task description): + * 3-node cluster — self at lsn=100, peer0 granted at lsn=99, peer1 granted at lsn=100 + * Sorted granted LSNs desc: [100 (self), 100 (peer1), 99 (peer0)] + * Majority threshold for 3 granted = ceil(3/2) = 2; element at index 1 (0-based) = 100 + * → safe_committed_lsn = 100 + */ +static void test_safe_committed_lsn_two_peers_at_99_and_100(void) +{ +#ifdef XCE_TESTING + /* n_other=2: ctx->votes[] has 2 entries */ + ha_elect_ctx_t *ctx = calloc(1, sizeof(*ctx) + 2 * sizeof(ctx->votes[0])); + TEST_ASSERT_NOT_NULL(ctx); + ctx->n_peers = 2; /* 2 other peers */ + ctx->votes_granted = 3; /* self + 2 peers granted */ + ctx->n_replied = 2; + pthread_mutex_init(&ctx->lock, NULL); + sem_init(&ctx->done, 0, 0); + + /* Self data stored in votes[0].my_lsn / my_tag */ + ctx->votes[0].my_lsn = 100; + ctx->votes[0].my_tag = 5; + ctx->votes[0].vote_granted = 1; /* peer0: granted, lsn=99 */ + ctx->votes[0].peer_lsn = 99; + ctx->votes[0].peer_tag = 5; + + ctx->votes[1].vote_granted = 1; /* peer1: granted, lsn=100 */ + ctx->votes[1].peer_lsn = 100; + ctx->votes[1].peer_tag = 5; + + uint64_t safe_lsn = 0, safe_tag = 0; + ha_compute_safe_committed_lsn(ctx, &safe_lsn, &safe_tag); + + TEST_ASSERT_EQUAL_UINT64_MESSAGE(100, safe_lsn, + "voters [99,100] + self=100 → safe_committed_lsn must be 100"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(5, safe_tag, + "voters all at tag=5 → safe_config_tag must be 5"); + + free_ctx(ctx); + printf("test_safe_committed_lsn_two_peers_at_99_and_100: PASS (safe_lsn=%" PRIu64 + ", safe_tag=%" PRIu64 ")\n", safe_lsn, safe_tag); +#else + printf("test_safe_committed_lsn_two_peers_at_99_and_100: SKIP (non-test build)\n"); +#endif +} + +/* + * Test case B (from task description): + * 3-node cluster — self at lsn=100, peer0 granted at lsn=50, peer1 granted at lsn=51 + * Sorted granted LSNs desc: [100 (self), 51 (peer1), 50 (peer0)] + * Majority threshold for 3 granted = 2; element at index 1 = 51 + * → safe_committed_lsn = 51 + */ +static void test_safe_committed_lsn_two_peers_at_50_and_51(void) +{ +#ifdef XCE_TESTING + ha_elect_ctx_t *ctx = calloc(1, sizeof(*ctx) + 2 * sizeof(ctx->votes[0])); + TEST_ASSERT_NOT_NULL(ctx); + ctx->n_peers = 2; + ctx->votes_granted = 3; + ctx->n_replied = 2; + pthread_mutex_init(&ctx->lock, NULL); + sem_init(&ctx->done, 0, 0); + + ctx->votes[0].my_lsn = 100; + ctx->votes[0].my_tag = 5; + ctx->votes[0].vote_granted = 1; /* peer0: granted, lsn=50 */ + ctx->votes[0].peer_lsn = 50; + ctx->votes[0].peer_tag = 5; + + ctx->votes[1].vote_granted = 1; /* peer1: granted, lsn=51 */ + ctx->votes[1].peer_lsn = 51; + ctx->votes[1].peer_tag = 5; + + uint64_t safe_lsn = 0, safe_tag = 0; + ha_compute_safe_committed_lsn(ctx, &safe_lsn, &safe_tag); + + TEST_ASSERT_EQUAL_UINT64_MESSAGE(51, safe_lsn, + "voters [50,51] + self=100 → safe_committed_lsn must be 51"); + + free_ctx(ctx); + printf("test_safe_committed_lsn_two_peers_at_50_and_51: PASS (safe_lsn=%" PRIu64 ")\n", + safe_lsn); +#else + printf("test_safe_committed_lsn_two_peers_at_50_and_51: SKIP (non-test build)\n"); +#endif +} + +/* + * Test case C: single-node cluster (n_peers=0, no votes[] entries). + * safe_committed_lsn must equal self's own LSN (self is the only majority). + */ +static void test_safe_committed_lsn_single_node(void) +{ +#ifdef XCE_TESTING + /* n_other=0: no votes[] entries; but we still need my_lsn from... somewhere. + * For the single-node case, ctx->n_peers=0 and votes[] is empty. + * The function should use 0 as safe_lsn (no entries to check). */ + ha_elect_ctx_t *ctx = calloc(1, sizeof(*ctx)); + TEST_ASSERT_NOT_NULL(ctx); + ctx->n_peers = 0; + ctx->votes_granted = 1; /* self-vote only */ + ctx->n_replied = 0; + pthread_mutex_init(&ctx->lock, NULL); + sem_init(&ctx->done, 0, 0); + + uint64_t safe_lsn = 0xDEAD, safe_tag = 0xDEAD; + ha_compute_safe_committed_lsn(ctx, &safe_lsn, &safe_tag); + + /* For single-node, safe_lsn should be 0 (no peer lsns to compare with) */ + TEST_ASSERT_EQUAL_UINT64_MESSAGE(0, safe_lsn, + "single-node (n_peers=0): safe_committed_lsn must be 0 (no peer data)"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(0, safe_tag, + "single-node (n_peers=0): safe_config_tag must be 0 (no peer data)"); + + free_ctx(ctx); + printf("test_safe_committed_lsn_single_node: PASS (safe_lsn=%" PRIu64 ")\n", safe_lsn); +#else + printf("test_safe_committed_lsn_single_node: SKIP (non-test build)\n"); +#endif +} + +/* + * Test case D: only one peer but it didn't grant vote. + * Only self vote is in the granted set; n_granted=1, majority threshold index=0. + * safe_lsn should be from the only granted voter (self), but since self's lsn + * is in votes[0].my_lsn, result should reflect that. + * This tests the partial-grant scenario. + */ +static void test_safe_committed_lsn_one_peer_denied(void) +{ +#ifdef XCE_TESTING + ha_elect_ctx_t *ctx = calloc(1, sizeof(*ctx) + 1 * sizeof(ctx->votes[0])); + TEST_ASSERT_NOT_NULL(ctx); + ctx->n_peers = 1; /* 1 other peer */ + ctx->votes_granted = 1; /* only self-vote granted */ + ctx->n_replied = 1; + pthread_mutex_init(&ctx->lock, NULL); + sem_init(&ctx->done, 0, 0); + + ctx->votes[0].my_lsn = 200; + ctx->votes[0].my_tag = 7; + ctx->votes[0].vote_granted = 0; /* peer denied vote */ + ctx->votes[0].peer_lsn = 150; + ctx->votes[0].peer_tag = 7; + + uint64_t safe_lsn = 0, safe_tag = 0; + ha_compute_safe_committed_lsn(ctx, &safe_lsn, &safe_tag); + + /* Only self granted (my_lsn=200). n_granted=1, majority index=0 → lsn=200 */ + TEST_ASSERT_EQUAL_UINT64_MESSAGE(200, safe_lsn, + "only self granted (my_lsn=200): safe_committed_lsn must be 200"); + + free_ctx(ctx); + printf("test_safe_committed_lsn_one_peer_denied: PASS (safe_lsn=%" PRIu64 ")\n", + safe_lsn); +#else + printf("test_safe_committed_lsn_one_peer_denied: SKIP (non-test build)\n"); +#endif +} + /* ── Suite entry point ───────────────────────────────────────────────────── */ void test_ha_evaluate_committing(void) @@ -441,4 +622,9 @@ void test_ha_evaluate_committing(void) RUN_TEST(test_done_posted_prevents_double_execute); RUN_TEST(test_committing_timeout_fn_promotes_to_primary); RUN_TEST(test_committing_timeout_fn_cancelled_is_noop); + /* Task 4.1 (ha-repl-hardening): safe committed LSN computation */ + RUN_TEST(test_safe_committed_lsn_two_peers_at_99_and_100); + RUN_TEST(test_safe_committed_lsn_two_peers_at_50_and_51); + RUN_TEST(test_safe_committed_lsn_single_node); + RUN_TEST(test_safe_committed_lsn_one_peer_denied); }