# Implementation Plan - [x] 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_ - [x] 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_ - [x] 3. Core: FSM gate extensions and role-based vote denial - [x] 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_ - [x] 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_ - [x] 4. Core: committing phase — replication push and availability gate - [x] 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_ - [x] 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_ - [x] 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_ - [x] 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_ - [x] 5. Integration: restructure myapp_run and wire all components - [x] 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_ - [x] 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_ - [x] 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_ - [x] 6. Validation: tests, ASAN, and integration verification - [x] 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_ - [x] 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_ - [x] 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_ - [x] 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_ - [x] 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.