myxce/.kiro/specs/election-iomgr-refactor/tasks.md

Implementation Plan

• 1. Foundation: delaymgr ms-precision and netio on_connected callback

• 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

• 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

• 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

• 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

• 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

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