From f0d6074d4eb6f2a9b57084ab72c99632e50c5cb3 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Fri, 12 Jun 2026 00:16:22 +0000 Subject: [PATCH] feat(ha-commit-engine): engine instantiation + sync-append path switch (task 5.3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit cluster_t embeds commit_engine_t; init in cluster_init_phase2 after the vtaskmgr guard (idempotence-guarded — phase2 can run twice), wiring repl_set_commit_engine + ha.repl back-pointer before any standby can register; fini after iomgr/vtask shutdown. persist_wal_sync_append passes written_lsn; engine-wired path drops -ENODEV and the committed_lsn store (engine owns advancement); LSN-0 first-entry routes to legacy per the accepted interim. p2 -ENODEV pin updated to the Req 6.3 contract. 34/34 + 6/6 green through honest delegation; census exactly at the 3 validated baselines; production binary links. --- .kiro/specs/ha-commit-engine/tasks.md | 6 +- .wolf/memory.md | 13 ++++ src/controller/cluster.c | 40 ++++++++++ src/controller/persist.c | 78 +++++++++++++------ .../testsuite/test_p2_integration.c | 53 +++++++------ src/include/xce/cluster_def.h | 2 + 6 files changed, 142 insertions(+), 50 deletions(-) diff --git a/.kiro/specs/ha-commit-engine/tasks.md b/.kiro/specs/ha-commit-engine/tasks.md index 7f3f92f..ff9e933 100644 --- a/.kiro/specs/ha-commit-engine/tasks.md +++ b/.kiro/specs/ha-commit-engine/tasks.md @@ -66,7 +66,7 @@ - _Depends: 2.2, 3.2_ - _Requirements: 1.1, 4.2_ -- [ ] 5.3 Implement engine instantiation and the sync-append path switch +- [x] 5.3 Implement engine instantiation and the sync-append path switch - cluster.h/cluster.c: embed `commit_engine_t commit` in `cluster_t`; call `commit_engine_init(&c->commit, &c->persist, n_peers, n_vcpus)` + `repl_set_commit_engine` at the init point where both the vtask pool size (cluster.c:1082 anchor) and `ha->n_peers` are final; set the `ha_state_t.repl` back-pointer here; `commit_engine_fini` on teardown - persist.c `persist_wal_sync_append`: pass `written_lsn` to the adapter; on the engine-wired path remove the `has_standbys → -ENODEV` special case and the `committed_lsn` store (persist.c:1035 anchor — the engine now owns advancement; truncation store at persist.c:1438 stays); return contract to callers: 0 / `-ETIMEDOUT` / `-XCE_ERR_NOT_READY` propagated through the existing job-submit error path - Observable completion: all task 5.1 tests PASS, including stale-ACK regression and single-node `committed_lsn` advancement; job submit on a wired cluster confirms only on per-entry majority @@ -95,4 +95,6 @@ - (5.2) commit.c linked into production controller_src ONE TASK EARLY (meson.build:3) — necessary (repl.c references engine symbols) and reviewer-verified inert (zero production callers of commit_engine_init/repl_set_commit_engine until 5.3; repl_init memsets ce=NULL → all new paths dormant). - (5.2) repl_broadcast's over-limit eviction (in_catch_up=1, repl.c:268-277) deliberately does NOT feed commit_on_membership_change — reviewer-proven no-op for waiter correctness (shrink can never raise the majority candidate; monotonic store). Design lists exactly four mutators; eviction recovery fires the hook at repl_complete_catchup. - (5.2) r->ce is read unlocked under a set-once-before-traffic contract (repl.h:47-52). 5.3 must wire it at cluster init BEFORE replication traffic; never NULL it concurrent with traffic (fini ordering is 5.3's responsibility). -- (5.2) Remaining RED going into 5.3: test 33 (Expected -110 Was 0 — persist passes entry_lsn=0 → legacy fallback) and test 34 (Expected 0 Was -19 — -ENODEV special case). These plus test 32's honest-delegation green are 5.3's completion evidence. +- (5.2) Remaining RED going into 5.3: test 33 (Expected -110 Was 0 — persist passes entry_lsn=0 → legacy fallback) and test 34 (Expected 0 Was -19 — -ENODEV special case). These plus test 32's honest-delegation green are 5.3's completion evidence. [RESOLVED at 5.3: both green] +- (5.3) For 6.1 audit: (a) baseline failure #1 (test_committed_lsn_does_not_advance_on_timeout) PHYSICALLY lives in test_p2_integration.c:277 — Unity misattributes it across groups (bug-411 artifact); its root cause is a stale expectation (old rc==0+sync_rc contract). (b) Legacy cluster_init() wires the engine with n_peers==0 (self-quorum) — confirmed test-only (zero production callers; production = myapp_run two-phase), but ce->n_peers is never re-derived if ha->n_peers is reassigned post-init (ha.c:3484 RDHUP re-election) — latent gap for future dynamic membership; consider an assert or re-derivation at the first-boot election callback. (c) Pre-existing teardown race (not 5.3's): persist_fini runs before vtaskmgr_fini in cluster_fini. +- (5.3) cluster_t lives in src/include/xce/cluster_def.h (not cluster.h); engine embedded there. Init point: cluster_init_phase2 after vtaskmgr guard, idempotence-guarded by c->commit.p==NULL (phase2 can run twice). n_vcpus = c->vtaskmgr.n_vcpus (resolved). diff --git a/.wolf/memory.md b/.wolf/memory.md index e84bb27..5b5d0b7 100644 --- a/.wolf/memory.md +++ b/.wolf/memory.md @@ -940,3 +940,16 @@ | 23:35 | ha-commit-engine 5.2: wired ce into replicator_t, ACK feed (repl_on_ack→commit_on_ack), 4 membership hooks, walAck delegation to commit_submit (gated ce!=NULL && entry_lsn!=0); added commit.c to controller_src for prod link | src/include/repl.h, src/controller/repl.c, src/controller/meson.build | tests 30-32 + 5 repl_majority walAck GREEN; 33/34 stay RED until 5.3; 3 baselines unchanged | ~45k | | 16:32 | Edited .kiro/specs/ha-commit-engine/tasks.md | inline fix | ~21 | | 16:32 | Edited .kiro/specs/ha-commit-engine/tasks.md | added error handling | ~345 | +| 16:43 | Edited src/include/xce/cluster_def.h | 2→3 lines | ~32 | +| 16:43 | Edited src/include/xce/cluster_def.h | 2→3 lines | ~59 | +| 16:44 | Edited src/controller/cluster.c | added 2 condition(s) | ~463 | +| 16:44 | Edited src/controller/cluster.c | added 1 condition(s) | ~170 | +| 16:45 | Edited src/controller/persist.c | added 2 condition(s) | ~869 | +| 16:45 | Edited src/controller/persist.c | 2→3 lines | ~20 | +| 16:45 | Edited src/controller/testsuite/test_p2_integration.c | modified test_single_node_committed_lsn_advances() | ~580 | +| 16:45 | Edited src/controller/testsuite/test_p2_integration.c | inline fix | ~15 | +| 16:45 | Edited src/controller/testsuite/test_p2_integration.c | 3→4 lines | ~21 | +| 23:58 | ha-commit-engine 5.3: engine instantiation (cluster_init_phase2, guard c->commit.p==NULL) + fini after iomgr_fini; persist_wal_sync_append passes written_lsn, engine path drops -ENODEV/committed_lsn store; p2 single-node pin updated to Req 6.3 contract | cluster_def.h, cluster.c, persist.c, test_p2_integration.c | 34/34 commit-engine GREEN, full census = 3 baselines only (repl_unit -110 line drifted 294->301 from p2 edit shift, value-identical) | ~60k | +| 00:25 | Adversarial review of ha-commit-engine task 5.3 (init wiring + sync-append switch): APPROVED — full census exactly 3 baseline fails, RED 32/34→34/34 verified via temp worktree at 6600fd5 | cluster.c, persist.c, cluster_def.h, test_p2_integration.c | approved | ~45k | +| 17:15 | Edited .kiro/specs/ha-commit-engine/tasks.md | inline fix | ~20 | +| 17:16 | Edited .kiro/specs/ha-commit-engine/tasks.md | modified race() | ~342 | diff --git a/src/controller/cluster.c b/src/controller/cluster.c index c1a440d..d913f72 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -1116,6 +1116,37 @@ int cluster_init_phase2(cluster_t* c) } } + /* ── Commit engine instantiation (ha-commit-engine task 5.3) ────────── + * Both inputs are final exactly here: n_vcpus was resolved by the + * vtaskmgr_init guard above (or by myapp_run before election) and + * ha->n_peers was fixed by ha_init_raft before the election that + * triggered this phase2. Legacy cluster_init() path (tests / + * single-node): ha->n_peers is still 0 → standalone self-quorum + * semantics, matching the old degraded single-node contract. + * + * Wire the engine BEFORE the promotion below: standbys register for + * replication only after the post-promotion ctrlhello, so r->ce is set + * once before any replication traffic (repl.h set-once contract). + * + * Guard c->commit.p: phase2 can run twice on the legacy wrapper path + * (cluster_init() directly, then the first-boot election callback); + * the engine must be initialized exactly once. */ + if (c->commit.p == NULL) { + rc = commit_engine_init(&c->commit, &c->persist, c->ha.n_peers, + c->vtaskmgr.n_vcpus); + if (rc != 0) { + ERROR("commit_engine_init failed: %d (n_peers=%d, n_vcpus=%d)", + rc, c->ha.n_peers, c->vtaskmgr.n_vcpus); + return rc; + } + repl_set_commit_engine(&c->persist.repl, &c->commit); + /* Write-admission gate reads registry truth through this + * back-pointer (ha-commit-engine task 4.2 / Req 3.1-3.3). */ + c->ha.repl = &c->persist.repl; + INFO("commit engine wired (n_peers=%d, n_vcpus=%d)", + c->ha.n_peers, c->vtaskmgr.n_vcpus); + } + /* Role set by ha_elect_evaluate_and_signal() inside cluster_run() */ if (ha_is_primary(&c->ha)) ha_promote_to_primary(c); @@ -1477,6 +1508,15 @@ void cluster_fini(cluster_t* c) iomgr_stop(&(c->iomgr)); /* join worker thread before freeing SSL_CTX */ iomgr_fini(&(c->iomgr)); + /* Commit engine teardown (ha-commit-engine task 5.3): destroy only after + * ALL replication traffic has stopped — vtaskmgr_fini joined the + * job-submit vtasks (commit_submit callers) and iomgr_stop/iomgr_fini + * joined the I/O thread (commit_on_ack source). r->ce is never NULLed + * while traffic lives (repl.h set-once contract); guard makes this safe + * when phase2 never ran (early init failure). */ + if (c->commit.p != NULL) + commit_engine_fini(&c->commit); + destroy(c->jobmgr); destroy(c->workermgr); destroy(c->qmgr); diff --git a/src/controller/persist.c b/src/controller/persist.c index 82444d5..e57c1bb 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -4,6 +4,7 @@ #include "ctrl_metrics.h" #include "hook.h" #include "xce/cluster_def.h" +#include "xce/errnox.h" #include "xce/types.h" #include "controller.capnp.h" #include "xlist.h" @@ -1011,44 +1012,71 @@ int persist_wal_sync_append(persist_t* p, uint32_t type, return rc; } - /* Majority-commit replication (Req 5.1–5.3). - * After local fsync via persist_wal_async, wait for majority quorum ACK. - * On quorum: advance committed_lsn. On no standbys or timeout: degraded. + /* Majority-commit replication (Req 5.1–5.3; ha-commit-engine Req 1.1–1.3). + * After local fsync via persist_wal_async, confirm per-entry majority. * We use current_lsn - 1 because persist_wal_async already incremented it. */ uint64_t written_lsn; pthread_mutex_lock(&p->lock); written_lsn = p->current_lsn - 1; pthread_mutex_unlock(&p->lock); - /* No standbys registered → degraded mode immediately (no quorum to wait for) */ - mylock_lock(&p->repl.lock); - int has_standbys = (p->repl.nstandbys > 0); - mylock_unlock(&p->repl.lock); - int sync_err; - if (!has_standbys) { - sync_err = -ENODEV; - } else { - /* entry_lsn = 0: mechanical signature update only (ha-commit-engine - * task 5.1) — the legacy body ignores it. Task 5.3 passes - * written_lsn so the wired engine evaluates per-entry majority. */ - sync_err = repl_wait_majority_ack(&p->repl, 0, p->config.sync_timeout_ms, + /* Engine-wired path (ha-commit-engine task 5.3): delegate the commit + * decision to the commit engine through the walAck adapter. The engine + * is the single committed_lsn writer (no store here), and its + * n_peers<=1 self-quorum replaces the old has_standbys → -ENODEV + * degraded special case (Req 6.3). + * + * LSN-0 edge (ACCEPTED interim behavior — spec Implementation Notes): + * written_lsn is passed unconditionally, but the FIRST-EVER WAL entry + * of a fresh cluster is LSN 0 and commit_submit rejects lsn==0, so the + * adapter's entry_lsn != 0 guard routes that one entry to the legacy + * branch below. This happens once per cluster lifetime; + * ha-single-leader-role's promotion no-op entry eliminates it in + * practice. */ + if (p->repl.ce != NULL && written_lsn != 0) { + sync_err = repl_wait_majority_ack(&p->repl, written_lsn, + p->config.sync_timeout_ms, (uint8_t)OPCode_walAck); - if (sync_err == 0) - atomic_store_explicit(&p->committed_lsn, written_lsn, memory_order_release); - else + if (sync_err != 0) CTRL_METRIC_INC(ctrl_repl_degraded_count); + } else { + /* DEPRECATED legacy branch: engine-less (ce == NULL) unit-test + * contexts plus the LSN-0 first entry. Connectivity-based counting; + * committed_lsn is stored here only because no engine owns it. + * Retired by ha-commit-barrier-rollout. */ + mylock_lock(&p->repl.lock); + int has_standbys = (p->repl.nstandbys > 0); + mylock_unlock(&p->repl.lock); + + if (!has_standbys) { + /* No standbys registered → degraded mode (no quorum to wait for) */ + sync_err = -ENODEV; + } else { + sync_err = repl_wait_majority_ack(&p->repl, written_lsn, + p->config.sync_timeout_ms, + (uint8_t)OPCode_walAck); + if (sync_err == 0) + atomic_store_explicit(&p->committed_lsn, written_lsn, + memory_order_release); + else + CTRL_METRIC_INC(ctrl_repl_degraded_count); + } } if (sync_rc_out) *sync_rc_out = sync_err; /* sync_err < 0 — replication failed. - * -ENODEV: no standbys registered (single-node or degraded). - * -ETIMEDOUT: standbys were registered but didn't ACK in time. - * Only -ENODEV is acceptable for forward progress; timeout means - * the cluster had standbys but couldn't reach quorum — the write - * must be rejected to preserve the majority-consistency invariant. */ - if (sync_err == -ETIMEDOUT) - return -ETIMEDOUT; + * -ENODEV: legacy degraded mode (no standbys, engine-less) — acceptable + * for forward progress. + * -ETIMEDOUT: no per-entry majority within the deadline — the write + * must be rejected to preserve the majority-consistency + * invariant. + * -XCE_ERR_NOT_READY: engine admission cap exhausted — retriable + * rejection. + * Timeout and admission rejection propagate to the job-submit error + * path (ha-commit-engine Req 1.1, 4.4). */ + if (sync_err == -ETIMEDOUT || sync_err == -XCE_ERR_NOT_READY) + return sync_err; return 0; } diff --git a/src/controller/testsuite/test_p2_integration.c b/src/controller/testsuite/test_p2_integration.c index d245748..506da47 100644 --- a/src/controller/testsuite/test_p2_integration.c +++ b/src/controller/testsuite/test_p2_integration.c @@ -18,6 +18,7 @@ #include +#include "commit.h" #include "config.h" #include "persist.h" #include "repl.h" @@ -219,44 +220,50 @@ static void test_repl_api_completeness(void) } /* ------------------------------------------------------------------ */ -/* Test 4b: Req 4.4 / single-node path — committed_lsn NOT advanced by - * persist_wal_sync_append when no standbys are registered. +/* Test 4b: single-node path — production contract (ha-commit-engine Req 6.3). * - * persist_wal_sync_append uses has_standbys flag. With no standbys it skips - * repl_wait_majority_ack entirely and sets sync_err = -ENODEV (degraded mode). - * The committed_lsn therefore remains unchanged. This is intentional: a - * single-node controller operates in degraded mode — WAL writes are locally - * durable but not majority-committed. Req 4.4 convergence only applies once - * standbys are registered and ACK. + * UPDATED by ha-commit-engine task 5.3: this test previously pinned the + * `has_standbys → -ENODEV` degraded special case, which 5.3 deliberately + * replaced. In production the commit engine is wired at cluster init, so a + * single-member primary (n_peers=1) self-commits: persist_wal_sync_append + * returns 0, sync status is 0 (NOT -ENODEV), and committed_lsn advances to + * the entry. The engine — not persist — owns the advancement. * - * This test documents the actual behaviour so regressions are caught if the - * single-node path is accidentally changed to advance committed_lsn. + * A priming async append makes the sync entry land at LSN 1: the first-ever + * WAL entry is LSN 0, which the adapter routes to the legacy branch (the + * accepted once-per-cluster-lifetime LSN-0 interim behaviour). */ -static void test_single_node_committed_lsn_unchanged(void) +static void test_single_node_committed_lsn_advances(void) { + commit_engine_t ce; + persist_t* p = make_p2_persist("clsn_1n"); TEST_ASSERT_NOT_NULL(p); - uint64_t committed_before = - atomic_load_explicit(&p->committed_lsn, memory_order_acquire); + /* Wire the engine exactly as cluster_init_phase2 does for n_peers=1 */ + TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&ce, p, 1, 8)); + repl_set_commit_engine(&p->repl, &ce); + /* Priming append: sync entry below lands at LSN 1 (engine-wired path) */ + TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "p", 1)); - int sync_rc = 0; + int sync_rc = -1; int rc = persist_wal_sync_append(p, WAL_JOB_SUBMIT, "job", 3, &sync_rc); /* WAL write must succeed locally */ TEST_ASSERT_EQUAL_INT(0, rc); - /* Degraded mode: no standbys → -ENODEV */ - TEST_ASSERT_EQUAL_INT(-ENODEV, sync_rc); - /* committed_lsn must remain at its pre-call value (not advanced) */ - uint64_t committed_after = - atomic_load_explicit(&p->committed_lsn, memory_order_acquire); - TEST_ASSERT_EQUAL_UINT64_MESSAGE(committed_before, committed_after, - "single-node: committed_lsn must stay unchanged (degraded, no standbys)"); + /* Single-member self-quorum: confirmed locally, no -ENODEV degraded mode */ + TEST_ASSERT_EQUAL_INT_MESSAGE(0, sync_rc, + "single-node wired engine must self-commit (0), not -ENODEV"); + /* committed_lsn advances to the entry (engine-owned advancement) */ + TEST_ASSERT_EQUAL_UINT64_MESSAGE( + 1, atomic_load_explicit(&p->committed_lsn, memory_order_acquire), + "single-node: committed_lsn must advance to the self-committed entry"); /* WAL LSN did advance (local write succeeded) */ TEST_ASSERT_TRUE(p->current_lsn > 0); + commit_engine_fini(&ce); teardown_p2_persist(p); - printf("test_single_node_committed_lsn_unchanged: PASS\n"); + printf("test_single_node_committed_lsn_advances: PASS\n"); } /* ------------------------------------------------------------------ */ @@ -407,7 +414,7 @@ void test_p2_integration(void) RUN_TEST(test_trim_lsn_with_active_standby_at_lsn_5); RUN_TEST(test_checkpoint_does_not_update_replicated_crc); RUN_TEST(test_sync_replication_degraded_mode); - RUN_TEST(test_single_node_committed_lsn_unchanged); + RUN_TEST(test_single_node_committed_lsn_advances); RUN_TEST(test_catchup_full_lifecycle_integration); RUN_TEST(test_needs_snapshot_boundary); RUN_TEST(test_repl_api_completeness); diff --git a/src/include/xce/cluster_def.h b/src/include/xce/cluster_def.h index 9198569..cbb6d71 100644 --- a/src/include/xce/cluster_def.h +++ b/src/include/xce/cluster_def.h @@ -18,6 +18,7 @@ #include "vtaskmgr.h" /* ha.h forward-declares cluster_t; including it here is safe (no circular dep) */ +#include "commit.h" #include "ha.h" #include "xce/types.h" #include "xlist.h" @@ -121,6 +122,7 @@ typedef struct cluster_s { int running; persist_t persist; /* Persistence module — P0-F */ ha_state_t ha; /* HA role management — P1-5 */ + commit_engine_t commit; /* Per-entry commit engine — ha-commit-engine 5.3 */ _Atomic int restart_requested; /* set by OP_upgradeConfig handler */ pthread_t heartbeat_listener_tid; /* pre-Phase-2 SSL listener thread */ _Atomic int heartbeat_listener_running;