diff --git a/.kiro/specs/ha-repl-hardening/tasks.md b/.kiro/specs/ha-repl-hardening/tasks.md index 363f0d34..a9600673 100644 --- a/.kiro/specs/ha-repl-hardening/tasks.md +++ b/.kiro/specs/ha-repl-hardening/tasks.md @@ -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 diff --git a/src/controller/ha.c b/src/controller/ha.c index 8ac6ebbb..00304acc 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -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); } diff --git a/src/controller/ha.h b/src/controller/ha.h index 86908be1..a44fecf3 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -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 diff --git a/src/controller/testsuite/test_ha_role.c b/src/controller/testsuite/test_ha_role.c index a34585b5..0a84298e 100644 --- a/src/controller/testsuite/test_ha_role.c +++ b/src/controller/testsuite/test_ha_role.c @@ -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) */