lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(ha-commit-engine): engine instantiation + sync-append path switch (task 5.3)

Browse source 
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.
This commit is contained in:
Rongsong Shen 2026-06-12 00:16:22 +00:00
parent 6600fd5d1f
commit f0d6074d4e
6 changed files with 142 additions and 50 deletions
Download patch file Download diff file Expand all files Collapse all files

40
src/controller/cluster.c
View file

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

78
src/controller/persist.c
View file

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

53
src/controller/testsuite/test_p2_integration.c
View file

@ -18,6 +18,7 @@
#include <unity/unity.h>
#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);

2
src/include/xce/cluster_def.h
View file

@ -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;