feat(election-iomgr-refactor): extend ha_dispatch_gate for CTRL_PRIMARY_COMMITTING — task 3.1

.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.

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) {

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);
 }