diff --git a/.kiro/specs/election-iomgr-refactor/tasks.md b/.kiro/specs/election-iomgr-refactor/tasks.md new file mode 100644 index 00000000..2df27258 --- /dev/null +++ b/.kiro/specs/election-iomgr-refactor/tasks.md @@ -0,0 +1,189 @@ +# Implementation Plan + +- [ ] 1. Foundation: delaymgr ms-precision and netio on_connected callback +- [x] 1.1 Add delay_task_ms() to delaymgr for millisecond-granularity one-shot tasks + - Add `struct timespec nextrun_ts` field to `delay_task_t` (alongside existing `time_t nextrun`) + - Add `delay_task_ms(mgr, delay_ms, runcount, fn, arg)` in delay.c using `clock_gettime(CLOCK_MONOTONIC)` for deadline; precision bounded by `mgr->precision_ms` + - Update `delay_worker` to check both `nextrun` (seconds tasks) and `nextrun_ts` (ms tasks) without breaking existing `delay_task` callers + - Expose `delay_task_ms` in delay.h + - Observable: `make test` passes existing delay tests; new unit test verifies a 500ms task fires within ±200ms + - _Requirements: 1.3, 4.4, 8.4_ + - _Boundary: delay.c, delay.h_ + +- [x] 1.2 Add on_connected callback to endpoint_t in netio + - Add `void (*on_connected)(endpoint_t *ep)` field to `endpoint_t` struct + - Add `endpoint_set_on_connected(endpoint_t *ep, void (*cb)(endpoint_t *))` helper function + - In `io_handler`, after `ssl_handshake_done` is set to 1, invoke `ep->on_connected(ep)` then null the pointer (fire-once semantics) + - Observable: unit test creates outgoing endpoint, sets callback, simulates handshake completion; callback fires exactly once + - _Requirements: 2.1, 2.3_ + - _Boundary: netio.c_ + +- [ ] 2. Core: iomgr-native outgoing vote connection and reply handling +- [x] 2.1 Add ha_state_t fields and CTRL_PRIMARY_COMMITTING to FSM enum + - Add to `ha_state_t`: `SSL_CTX *client_ssl_ctx`, `int phase2_done`, `_Atomic int repl_ack_count`, `int n_peers` + - Add to `ha_config_t`: `uint32_t committing_timeout_ms` with default `HA_DEFAULT_COMMITTING_TIMEOUT_MS = 10000` + - Add `CTRL_PRIMARY_COMMITTING` to `ctrl_role_t` enum in ha.h + - Initialize `phase2_done=0`, `repl_ack_count=0` in `ha_init()` + - Observable: project builds without errors; `ha_init` zero-initializes all new fields + - _Requirements: 5.2, 8.1_ + - _Boundary: ha.h_ + +- [x] 2.2 (P) Implement vote_ep_ctx_t and on_vote_connected callback + - Define `vote_ep_ctx_t` struct: `ha_state_t *ha`, `cluster_t *c`, `ha_elect_ctx_t *ctx`, `int peer_idx`, `uint64_t my_term/my_tag/my_lsn`, `const char *my_name` + - Implement `on_vote_connected(endpoint_t *ep)`: cast `ep->data` to `vote_ep_ctx_t`, build `xce_request_t{opcode=OPCode_requestVote, body.requestvotemsg=...}`, call `xce_request(ep, &req)`, set `ep->msg_handler = vote_reply_msg_handler` + - Define `election_timeout_ctx_t`: `_Atomic int cancelled`, `ha_state_t *ha`, `cluster_t *c`, `ha_elect_ctx_t *ctx` + - Observable: unit test with mock endpoint verifies `xce_request` called with correct opcode and fields after `on_vote_connected` fires + - _Requirements: 1.2, 2.3, 3.1_ + - _Boundary: ha.c (vote send path)_ + - _Depends: 1.2, 2.1_ + +- [x] 2.3 (P) Implement vote_reply_msg_handler for iomgr-native reply decoding + - Implement `vote_reply_msg_handler(endpoint_t *ep)`: call `xce_decode(pkt, &req)`, verify `req.opcode == OPCode_reply`, extract `rep->data.requestvotereplymsg`, call `read_RequestVoteReply(&vr_raw, ...)` BEFORE any capnp_free, call `handle_request_vote_reply(ha, ctx, peer_idx, vr_raw.term, vr_raw.voteGranted, ...)` + - Term fencing: if `vr_raw.term > ha->term`, atomically update `ha->term` before recording vote + - Observable: ASAN-clean unit test reads reply fields before free; no heap-use-after-free + - _Requirements: 3.1, 3.2, 3.3, 3.4_ + - _Boundary: ha.c (vote reply path)_ + - _Depends: 2.1_ + +- [x] 2.4 Implement ha_start_election_async() — non-blocking election initiator + - Implement `ha_start_election_async(ha, c, n_peers, peers, my_config_tag, my_lsn, my_name)`: + - Store `n_peers` in `ha->n_peers` + - For each peer: allocate `vote_ep_ctx_t`, init endpoint with `client_ssl_ctx`, set `on_vote_connected` + `vote_ep_err_handler`, call `endpoint_connect` + `iomgr_monitor`; skip peer on connect failure (counted as deny via error handler) + - Allocate `election_timeout_ctx_t` with `cancelled=0`; schedule `delay_task_ms(delaymgr, election_timeout_ms, 1, election_timeout_fn, ctx)` + - Single-node shortcircuit: if n_peers==1, call `ha_elect_evaluate_and_signal` directly with PRIMARY result + - Returns immediately; election result delivered via `ha_elect_evaluate_and_signal` + - Observable: function returns in < 5ms; endpoints registered in iomgr; watchdog scheduled in delaymgr + - _Requirements: 1.1, 1.2, 1.3, 2.1, 2.4, 4.3_ + - _Boundary: ha.c (election orchestration)_ + - _Depends: 1.1, 1.2, 2.2, 2.3_ + +- [ ] 3. Core: FSM gate extensions and role-based vote denial +- [ ] 3.1 (P) Extend ha_dispatch_gate for CTRL_PRIMARY_COMMITTING state + - Add `case CTRL_PRIMARY_COMMITTING:` to `ha_dispatch_gate` in ha.c: allow `OPCode_heartbeat`, `OPCode_replConfig`, `OPCode_replConfigPush`, `OPCode_upgradeConfigAck`, `OPCode_hello`, `OPCode_walAck`, `OPCode_replEntry`, `OPCode_replAck`, `OPCode_replCatchUpDone`; deny all others with `-XCE_ERR_NOT_READY` + - Observable: unit test sends `OPCode_jobsubmit` in COMMITTING state → returns `XCE_ERR_NOT_READY`; `OPCode_replCatchUpDone` → returns 0 + - _Requirements: 5.1, 5.2, 8.1_ + - _Boundary: ha.c (dispatch gate)_ + - _Depends: 2.1_ + +- [ ] 3.2 (P) Extend handle_request_vote with role-based deny gate + - Before existing term/voted_for/freshness checks in `handle_request_vote`: load `ha->role`; return 0 (deny) if `CTRL_PRIMARY` or `CTRL_PRIMARY_COMMITTING`; return 0 (deny) if `CTRL_STANDBY && ha_heartbeat_is_fresh(ha)` + - Implement `ha_heartbeat_is_fresh(ha)`: compare `clock_gettime(CLOCK_MONOTONIC)` against `ha->last_heartbeat`; return 1 if elapsed < `heartbeat_timeout_ms` + - Observable: unit test with mocked roles: PRIMARY→deny, COMMITTING→deny, STANDBY+fresh→deny, STANDBY+stale→evaluate, ELECTING→evaluate + - _Requirements: 6.1, 6.2, 6.4_ + - _Boundary: ha.c (vote gate)_ + - _Depends: 2.1_ + +- [ ] 4. Core: committing phase — replication push and availability gate +- [ ] 4.1 Extend ha_elect_evaluate_and_signal with COMMITTING path + - Extend `ha_elect_evaluate_and_signal`: after determining role=PRIMARY, check `!ha->phase2_done`; if first boot call `cluster_init_phase2(c)` then set `ha->phase2_done=1`; set `ha->role = CTRL_PRIMARY_COMMITTING`; reset `repl_ack_count=0`; schedule committing timeout watchdog via `delay_task_ms` + - On STANDBY result: if first boot call `cluster_init_phase2(c)` + set `phase2_done=1`; set role=CTRL_STANDBY; election timeout callback sets `cancelled=1` + - Observable: unit test verifies `cluster_init_phase2` called exactly once on first election; second election (RDHUP) skips it; role enters COMMITTING not PRIMARY after winning + - _Requirements: 4.1, 5.2, 9.2_ + - _Boundary: ha.c (election evaluation)_ + - _Depends: 2.4, 3.1_ + +- [ ] 4.2 Implement config and WAL replication push to standbys + - In `ha_elect_evaluate_and_signal` COMMITTING path: call `push_config_to_standbys(c, ...)` to push config to standbys with lower `config_tag` + - Implement `ha_send_ctrlhello_to_all_standbys(c)`: iterate `ha->standby_endpoints`, send `CtrlHelloRequest` to each via `xce_request` — triggers existing catch-up replication in `handle_ctrlhello` + - Observable: 3-node startup log shows primary pushing config to ctrl-b/c immediately after COMMITTING; ctrl-b/c logs show config received + - _Requirements: 7.1, 7.2, 7.3_ + - _Boundary: ha.c + cluster.c (replication push)_ + - _Depends: 4.1_ + +- [ ] 4.3 Implement ha_repl_ack_received and majority-ack → PRIMARY transition + - Implement `ha_repl_ack_received(ha, c)`: atomically increment `repl_ack_count`; compute quorum `= (ha->n_peers + 2) / 2 - 1`; if `repl_ack_count >= quorum`: set committing timeout `cancelled=1`, call `ha_promote_to_primary(c)` (sets role=CTRL_PRIMARY, enables hooks) + - Wire `ha_repl_ack_received` into `OPCode_replCatchUpDone` handler in cluster.c + - Observable: unit test: 2 of 3 acks arrive → PRIMARY; 1 of 3 acks → stays COMMITTING + - _Requirements: 7.4, 8.2, 8.3_ + - _Boundary: ha.c + cluster.c (ack counting)_ + - _Depends: 4.1_ + +- [ ] 4.4 Implement committing timeout watchdog (degraded-mode fallback) + - Define `committing_timeout_ctx_t`: `_Atomic int cancelled`, `ha_state_t *ha`, `cluster_t *c` + - Implement `committing_timeout_fn`: if `!cancelled` and `ha->role == CTRL_PRIMARY_COMMITTING`, log WARN "committing timeout — opening in degraded mode", call `ha_promote_to_primary(c)` + - Schedule via `delay_task_ms(delaymgr, committing_timeout_ms, 1, committing_timeout_fn, ctx)` in the COMMITTING entry path + - Observable: unit test with no standbys: after `committing_timeout_ms`, role transitions to CTRL_PRIMARY with WARN log + - _Requirements: 8.4, 9.1_ + - _Boundary: ha.c (committing watchdog)_ + - _Depends: 1.1, 4.1_ + +- [ ] 5. Integration: restructure myapp_run and wire all components +- [ ] 5.1 Restructure myapp_run to be fully non-blocking + - Move `delaymgr_init` + `delaymgr_start` and `persist_init` + `persist_start` from `cluster_start` path into `myapp_run` (before `ha_start_election_async`) + - Create `client_ssl_ctx` (TLS_client_method + SSL_VERIFY_NONE) in `myapp_run`; store in `c.ha.client_ssl_ctx` + - Call `load_primary_config` to pre-load config_tag for freshness gate + - Remove: `ha_wait_for_heartbeat_or_timeout`, `am_i_primary_quorum`, `for(;;)` retry loop, `cluster_init_phase2`, `cluster_start`, `bootstrap_config_if_needed` calls + - Call `ha_start_election_async` then `cluster_run` directly — no blocking between them + - Observable: `myapp_run` exits into `cluster_run` within 10ms of startup; no blocking calls visible in strace/callgraph + - _Requirements: 1.1, 9.3, 9.4_ + - _Boundary: myapp.c_ + - _Depends: 2.4, 3.1, 3.2, 4.1, 4.2, 4.3, 4.4_ + +- [ ] 5.2 Wire heartbeat-timeout → CTRL_ELECTING → ha_start_election_async for RDHUP failover + - In `ha_on_primary_ctrl_fd_rdhup`: after setting role=CTRL_ELECTING, call `ha_start_election_async` with the cluster's peer list (instead of the current `am_i_primary_quorum` call) + - Verify `ha->phase2_done==1` ensures RDHUP re-election skips `cluster_init_phase2` and goes directly to COMMITTING + - Observable: kill primary with SIGKILL; surviving standby transitions ELECTING→COMMITTING→PRIMARY within `heartbeat_timeout_ms + election_timeout_ms + committing_timeout_ms` + - _Requirements: 6.3, 6.5, 9.3_ + - _Boundary: ha.c (RDHUP path)_ + - _Depends: 5.1_ + +- [ ] 5.3 Remove bootstrap_config_if_needed and old election blocking path + - Delete `bootstrap_config_if_needed` call from myapp_run; delete the function body from config_repl.c if no other callers remain + - Delete `ha_wait_for_heartbeat_or_timeout` call from myapp_run (now internal to ha_start_election_async's watchdog) + - Delete `am_i_primary_quorum` blocking pthread loop from ha.c (keep unit-test stub in testsuite) + - Delete `vote_request_thread` and `vote_thread_ctx_t` from ha.c + - Observable: `grep -r "bootstrap_config_if_needed\|vote_request_thread\|vote_thread_ctx_t" src/controller/ --include="*.c"` returns no matches outside of test stubs + - _Requirements: 1.1, 9.2_ + - _Boundary: ha.c, myapp.c, config_repl.c_ + - _Depends: 5.1, 5.2_ + +- [ ] 6. Validation: tests, ASAN, and integration verification +- [ ] 6.1 Unit tests for election timeout and committing timeout watchdogs + - Test `election_timeout_fn`: fires `ha_elect_evaluate_and_signal` when not cancelled; no-op when `cancelled=1` + - Test `committing_timeout_fn`: promotes to PRIMARY (degraded) when not cancelled; no-op when cancelled + - Test `delay_task_ms` precision: 500ms task fires within ±200ms (delaymgr precision_ms = 50ms for test) + - Observable: all tests pass; no ASAN errors + - _Requirements: 1.3, 4.3, 8.4_ + - _Boundary: ha.c unit tests_ + - _Depends: 1.1, 4.4_ + +- [ ] 6.2 Unit tests for FSM gate and role-based vote denial + - Test `ha_dispatch_gate`: COMMITTING allows replCatchUpDone, denies jobsubmit; ELECTING allows requestVote, denies jobsubmit + - Test `handle_request_vote` role gate: all 5 cases (PRIMARY, COMMITTING, STANDBY+fresh, STANDBY+stale, ELECTING) + - Test `ha_repl_ack_received`: quorum=1 of 2 peers acks → PRIMARY; 0 of 2 → stays COMMITTING + - Observable: 15+ new unit test cases pass; `make test` green + - _Requirements: 5.1, 5.2, 6.1, 6.2, 6.4, 8.2, 8.3_ + - _Boundary: ha.c unit tests, cluster.c unit tests_ + - _Depends: 3.1, 3.2, 4.3_ + +- [ ] 6.3 Integration test: 3-node startup with full COMMITTING phase + - Start ctrl-a, ctrl-b, ctrl-c within 3 seconds; verify ctrl-a wins election and enters COMMITTING + - Verify ctrl-b/c receive config push (log: "config received from primary"); both send replCatchUpDone + - Verify ctrl-a transitions to CTRL_PRIMARY after 2 acks; verify job dispatch accepted after gate opens + - Verify ctrl-b/c deny vote requests while ctrl-a is COMMITTING or PRIMARY (role gate active) + - Observable: cluster serves a job within 30s of all nodes starting; no XCE_ERR_NOT_READY from primary after COMMITTING exits + - _Requirements: 4.1, 7.1, 7.2, 7.3, 8.1, 8.2, 8.3_ + - _Depends: 5.1, 5.2, 5.3_ + +- [ ] 6.4 Integration test: primary crash and standby promotion (RDHUP failover) + - Start 3-node cluster; let ctrl-a become PRIMARY; SIGKILL ctrl-a + - Verify ctrl-b/c detect heartbeat timeout within `heartbeat_timeout_ms`; one wins election → COMMITTING → PRIMARY + - Verify promoted primary serves job traffic from last WAL state (no data loss) + - Verify ASAN-clean run: no heap-use-after-free, no stack-use-after-return + - Observable: cluster operational again within `heartbeat_timeout_ms + election_timeout_ms + committing_timeout_ms + 5s`; ASAN output shows 0 errors + - _Requirements: 6.3, 6.5, 9.3_ + - _Depends: 6.3_ + +- [ ] 6.5 Regression: existing unit test suite green; ASAN clean + - Run full `meson test --suite controller` with `ASAN_OPTIONS=detect_leaks=0`; all 230+ existing tests pass + - Verify single-node self-election still works without any network I/O + - Verify no compilation warnings introduced in modified files + - Observable: `meson test --suite controller` exits 0; test count ≥ prior baseline; no new ASAN errors + - _Requirements: 9.1, 9.4_ + - _Depends: 6.4_ + +## Implementation Notes + +- `ha_elect_evaluate_and_signal` is called from vtask context (via vote_reply_msg_handler) and from delaymgr context (via election_timeout_fn). Both are thread-safe because `ha_elect_ctx_t` operations are mutex-protected and `ha->role` is atomic. +- `cluster_init_phase2` must be verified vtask-safe before task 4.1: check that all subsystems it touches (configmgr, jobmgr, etc.) were moved to phase1 in the prior refactor. +- The committing timeout and election timeout contexts must outlive the callbacks. Allocate them on heap inside `ha_start_election_async`; free in `ha_election_cleanup` called after COMMITTING exits. diff --git a/src/controller/ha.c b/src/controller/ha.c index 79f37709..eb3d2704 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -511,6 +511,26 @@ int ha_dispatch_gate(ha_state_t *ha, int opcode) return 0; return -XCE_ERR_NOT_READY; + case CTRL_PRIMARY_COMMITTING: + /* Req 5.1, 5.2, 8.1: Primary won election but has not yet received + * majority replication ack. Allow only replication-related and + * administrative opcodes; deny job dispatch and queue management + * with XCE_ERR_NOT_READY. */ + switch (opcode) { + case OPCode_heartbeat: + case OPCode_replConfig: + case OPCode_replConfigPush: + case OPCode_upgradeConfigAck: + case OPCode_hello: + case OPCode_walAck: + case OPCode_replEntry: + case OPCode_replAck: + case OPCode_replCatchUpDone: + return 0; + default: + return -XCE_ERR_NOT_READY; + } + case CTRL_STANDBY: { /* Allow ops that standbys legitimately handle */ switch (opcode) { diff --git a/src/controller/testsuite/test_ha_startup.c b/src/controller/testsuite/test_ha_startup.c index ba28718c..9bb1da24 100644 --- a/src/controller/testsuite/test_ha_startup.c +++ b/src/controller/testsuite/test_ha_startup.c @@ -540,6 +540,49 @@ static void test_dispatch_gate_init_allows_heartbeat(void) printf("test_dispatch_gate_init_allows_heartbeat: PASS\n"); } +/* ── Task 3.1: dispatch gate CTRL_PRIMARY_COMMITTING ─────────────────────── */ + +static void test_dispatch_gate_committing_denies_job_ops(void) +{ + /* Req 5.1, 8.1: When role is CTRL_PRIMARY_COMMITTING, primary-only + * job operations must be denied with XCE_ERR_NOT_READY. + * OPCode_jobsub is a representative job-submission opcode. */ + ha_state_t ha; + ha_init(&ha, &g_dispatch_cfg, 0); + atomic_store_explicit(&ha.role, (int32_t)CTRL_PRIMARY_COMMITTING, + memory_order_release); + + int rc = ha_dispatch_gate(&ha, OPCode_jobsub); + TEST_ASSERT_EQUAL_INT(-XCE_ERR_NOT_READY, rc); + + ha_fini(&ha); + printf("test_dispatch_gate_committing_denies_job_ops: PASS\n"); +} + +static void test_dispatch_gate_committing_allows_repl_ops(void) +{ + /* Req 5.2, 8.1: When role is CTRL_PRIMARY_COMMITTING, replication + * operations must be permitted (return 0). + * OPCode_replCatchUpDone is the key signal that ends the COMMITTING phase. */ + ha_state_t ha; + ha_init(&ha, &g_dispatch_cfg, 0); + atomic_store_explicit(&ha.role, (int32_t)CTRL_PRIMARY_COMMITTING, + memory_order_release); + + TEST_ASSERT_EQUAL_INT(0, ha_dispatch_gate(&ha, OPCode_replCatchUpDone)); + TEST_ASSERT_EQUAL_INT(0, ha_dispatch_gate(&ha, OPCode_heartbeat)); + TEST_ASSERT_EQUAL_INT(0, ha_dispatch_gate(&ha, OPCode_replConfig)); + TEST_ASSERT_EQUAL_INT(0, ha_dispatch_gate(&ha, OPCode_replConfigPush)); + TEST_ASSERT_EQUAL_INT(0, ha_dispatch_gate(&ha, OPCode_upgradeConfigAck)); + TEST_ASSERT_EQUAL_INT(0, ha_dispatch_gate(&ha, OPCode_hello)); + TEST_ASSERT_EQUAL_INT(0, ha_dispatch_gate(&ha, OPCode_walAck)); + TEST_ASSERT_EQUAL_INT(0, ha_dispatch_gate(&ha, OPCode_replEntry)); + TEST_ASSERT_EQUAL_INT(0, ha_dispatch_gate(&ha, OPCode_replAck)); + + ha_fini(&ha); + printf("test_dispatch_gate_committing_allows_repl_ops: PASS\n"); +} + /* ── Test suite entry point ───────────────────────────────────────────── */ void test_ha_startup(void) @@ -587,4 +630,8 @@ void test_ha_startup(void) /* Req 1.1 — dispatch gate allows heartbeat in CTRL_INIT */ RUN_TEST(test_dispatch_gate_init_allows_heartbeat); + + /* Task 3.1 — dispatch gate: CTRL_PRIMARY_COMMITTING */ + RUN_TEST(test_dispatch_gate_committing_denies_job_ops); + RUN_TEST(test_dispatch_gate_committing_allows_repl_ops); }