From eae4767d7d86fd0aa252346f18f869370919e33d Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 11 Jun 2026 14:37:01 +0000 Subject: [PATCH] feat(ha-commit-engine): per-LSN majority commit engine core (task 1.2) commit_on_ack: active-standby (non-catch-up, connected) watermark scan, descending sort, candidate arr[n_peers/2-1], clamped to appended log, monotonic single-site committed_lsn store, condvar broadcast, never blocks beyond engine mutex. RETAIN returns immediately. Self-quorum for n_peers<=1; zero-fill makes zero-standby commit impossible. 7/7 task-1.1 tests green unmodified; full suite at known 3-failure baseline. --- .kiro/specs/ha-commit-engine/tasks.md | 2 +- .wolf/anatomy.md | 4 +- .wolf/memory.md | 3 + src/controller/commit.c | 217 +++++++++++++++++++++++--- 4 files changed, 203 insertions(+), 23 deletions(-) diff --git a/.kiro/specs/ha-commit-engine/tasks.md b/.kiro/specs/ha-commit-engine/tasks.md index 99eb9dcb..ebd444d9 100644 --- a/.kiro/specs/ha-commit-engine/tasks.md +++ b/.kiro/specs/ha-commit-engine/tasks.md @@ -10,7 +10,7 @@ - Observable completion: build succeeds with the new header and test file; all new test cases FAIL (red) against a stub `commit.c` or unlinked implementation - _Requirements: 1.2, 2.1, 2.2, 2.3, 5.1_ -- [ ] 1.2 Implement the commit engine core to green +- [x] 1.2 Implement the commit engine core to green - Create `src/controller/commit.c` (add to meson controller sources): `commit_engine_init`/`commit_engine_fini` (validate args, one mutex + one condvar), `commit_on_ack` (bounded scan of active standbys' `acked_lsn` into the scratch array, sort descending, candidate = `arr[n_peers/2 - 1]` clamped to appended log, monotonic `max()` store into `persist->committed_lsn`, condvar broadcast — never blocks), `commit_submit(RETAIN)` immediate return - Catch-up standbys (`in_catch_up`) excluded from the evaluation count (umbrella-locked conservative rule) - XCE_TESTING asserts for monotonicity diff --git a/.wolf/anatomy.md b/.wolf/anatomy.md index cb660387..1a1d32a6 100644 --- a/.wolf/anatomy.md +++ b/.wolf/anatomy.md @@ -1,6 +1,6 @@ # anatomy.md -> Auto-maintained by OpenWolf. Last scanned: 2026-06-11T14:16:52.677Z +> Auto-maintained by OpenWolf. Last scanned: 2026-06-11T14:36:56.598Z > Files: 511 tracked | Anatomy hits: 0 | Misses: 0 ## ./ @@ -706,7 +706,7 @@ ## src/controller/ -- `commit.c` — STUB commit engine (ha-commit-engine task 1.1 RED): init/fini real, commit_submit -ENOSYS, commit_on_ack no-op; real impl in task 1.2 (~474 tok) +- `commit.c` — Commit engine core (ha-commit-engine 1.2): per-LSN majority evaluation (descending watermark median arr[n_peers/2-1], catch-up excluded), monotonic clamped committed_lsn advancement, RETAIN submit immediate; WAIT = -ENOSYS until task 2.2 (~2600 tok) - `meson.build` (~1094 tok) ## src/controller/include/ diff --git a/.wolf/memory.md b/.wolf/memory.md index d84bca56..c425dd2a 100644 --- a/.wolf/memory.md +++ b/.wolf/memory.md @@ -870,3 +870,6 @@ | 13:30 | Task 1.1 ha-commit-engine RED: commit.h + stub commit.c + test_commit_engine.c (7 failing tests) | src/controller/{include/commit.h,commit.c,testsuite/test_commit_engine.c,meson.build,testsuite/controller_test.c} | 7/7 RED for right reasons; 3 suite failures proven pre-existing via baseline stash run | ~60k | | 07:16 | Edited .kiro/specs/ha-commit-engine/tasks.md | inline fix | ~23 | | 07:16 | Edited .kiro/specs/ha-commit-engine/tasks.md | modified now() | ~344 | +| 07:22 | Created src/controller/commit.c | — | ~2108 | +| 2026-06-11 | ha-commit-engine task 1.2: implemented commit engine core (majority median, monotonic clamped committed_lsn, RETAIN) — 7/7 tests green | src/controller/commit.c | success | ~9k | +| 07:36 | Edited .kiro/specs/ha-commit-engine/tasks.md | inline fix | ~14 | diff --git a/src/controller/commit.c b/src/controller/commit.c index 4efda5a7..147367d9 100644 --- a/src/controller/commit.c +++ b/src/controller/commit.c @@ -1,17 +1,148 @@ /* - * commit.c — STUB for ha-commit-engine task 1.1 (RED phase). + * commit.c — Commit engine core (ha-commit-engine task 1.2). * - * Behavioral functions are deliberately inert (-ENOSYS / no-op) so the - * task 1.1 tests in testsuite/test_commit_engine.c compile, link, and FAIL - * on their assertions. init/fini are real enough that tests exercise the - * engine's behavioral contract rather than failing at fixture setup. + * Single authority for per-entry commit evaluation and committed_lsn + * advancement (Raft §5.3, umbrella-verbatim commit rule): * - * The real implementation lands in task 1.2 (per-LSN majority evaluation, - * monotonic/contiguous committed_lsn advancement). + * entry N is committed when + * count(self + active standbys with acked_lsn >= N) > n_peers / 2 + * + * Evaluated as a watermark median: active (connected, non-catch-up) + * standby watermarks are scanned into a scratch array, sorted descending, + * and the candidate is arr[n_peers/2 - 1] — the highest LSN covered by + * self plus a majority of peers. The candidate is clamped to the highest + * locally appended LSN and stored with monotonic max() semantics. + * + * commit_on_ack is called from the iomgr read path and never blocks + * beyond the engine mutex (plus the registry slot mutex, which the same + * iomgr path already takes in repl_on_ack): no I/O, no condvar waits. + * + * COMMIT_MODE_WAIT (blocking wait + admission cap) lands in task 2.2. + * + * Requirements: 1.2, 2.1, 2.2, 2.3, 5.1 */ #include +#include +#include #include "commit.h" +#include "mylock.h" +#include "utils.h" + +#ifdef XCE_TESTING +#include +#endif + +/* ============================================================ + * Internal helpers + * ============================================================ */ + +/* Highest locally appended WAL LSN. persist invariant: + * committed_lsn <= current_lsn - 1 (current_lsn is the NEXT LSN). */ +static uint64_t highest_appended_lsn(const persist_t* p) +{ + return (p->current_lsn > 0) ? p->current_lsn - 1 : 0; +} + +/* Descending insertion sort — n is bounded by MAX_STANDBYS. */ +static void sort_desc(uint64_t* arr, int n) +{ + int i, j; + uint64_t v; + + for (i = 1; i < n; i++) { + v = arr[i]; + for (j = i - 1; j >= 0 && arr[j] < v; j--) { + arr[j + 1] = arr[j]; + } + arr[j + 1] = v; + } +} + +/* + * Compute the highest LSN covered by self + a majority of peers. + * Caller holds ce->lock. + * + * Catch-up standbys are excluded from the evaluation count + * (umbrella-locked conservative rule); absent peers count as watermark 0. + */ +static uint64_t majority_candidate_locked(commit_engine_t* ce) +{ + replicator_t* r; + uint64_t cand; + uint64_t cap = highest_appended_lsn(ce->p); + int n = 0; + int need, k, i; + + /* Standalone / single-member: self constitutes the majority (Req 6.3). */ + if (ce->n_peers <= 1) { + return cap; + } + + /* Bounded scan of active standby watermarks into the scratch array. */ + r = &ce->p->repl; + mylock_lock(&r->lock); + for (i = 0; i < MAX_STANDBYS && n < MAX_STANDBYS; i++) { + standby_conn_t* s = &r->standbys[i]; + + if (s->ep == NULL || atomic_load(&s->in_catch_up) != 0) { + continue; /* empty slot or catch-up: not countable toward majority */ + } + ce->scratch[n++] = atomic_load(&s->acked_lsn); + } + mylock_unlock(&r->lock); + + /* Zero-fill up to the full peer count so missing peers count as 0. */ + need = ce->n_peers - 1; + if (need > MAX_STANDBYS) { + need = MAX_STANDBYS; + } + while (n < need) { + ce->scratch[n++] = 0; + } + + /* Self always covers every appended entry, so a majority needs + * n_peers/2 standbys at-or-above N; sorted descending, the highest + * such N is the watermark at index n_peers/2 - 1. */ + k = ce->n_peers / 2 - 1; + if (k >= n) { + return 0; /* peer count not representable — no majority possible */ + } + sort_desc(ce->scratch, n); + cand = ce->scratch[k]; + + /* Contiguous and bounded: never beyond the appended log (Req 2.3). */ + return (cand < cap) ? cand : cap; +} + +/* + * Re-evaluate majority and advance committed_lsn (monotonic max() store). + * Caller holds ce->lock. Broadcasts the condvar on advancement so future + * WAIT-mode blockers (task 2.2) re-check their per-LSN predicates. + */ +static void commit_advance_locked(commit_engine_t* ce) +{ + uint64_t cand = majority_candidate_locked(ce); + uint64_t cur = atomic_load(&ce->p->committed_lsn); + + if (cand <= cur) { + return; /* monotonic: committed_lsn never regresses (Req 2.2) */ + } + +#ifdef XCE_TESTING + assert(cand > cur); /* monotonic advance only */ + assert(cand <= highest_appended_lsn(ce->p)); /* never beyond the log */ +#endif + + atomic_store(&ce->p->committed_lsn, cand); + TRACE("commit: committed_lsn %lu -> %lu (n_peers=%d)", cur, cand, + ce->n_peers); + pthread_cond_broadcast(&ce->cond); +} + +/* ============================================================ + * Lifecycle + * ============================================================ */ int commit_engine_init(commit_engine_t* ce, persist_t* p, int n_peers, int n_vcpus) { @@ -24,7 +155,8 @@ int commit_engine_init(commit_engine_t* ce, persist_t* p, int n_peers, int n_vcp ce->n_peers = n_peers; ce->n_vcpus = n_vcpus; ce->inflight_waits = 0; - ce->writable_lsn = 0; + ce->writable_lsn = 0; /* no own-term gate until set by promotion */ + memset(ce->scratch, 0, sizeof(ce->scratch)); return 0; } @@ -38,31 +170,76 @@ void commit_engine_fini(commit_engine_t* ce) ce->p = NULL; } +/* ============================================================ + * Entry registration + * ============================================================ */ + int commit_submit(commit_engine_t* ce, uint64_t lsn, commit_mode_t mode, uint32_t timeout_ms) { - (void)ce; - (void)lsn; - (void)mode; - (void)timeout_ms; - return -ENOSYS; /* stub: real WAIT/RETAIN semantics in task 1.2/2.2 */ + (void)timeout_ms; /* RETAIN never honors the timeout; WAIT uses it (2.2) */ + + if (ce == NULL || ce->p == NULL || lsn == 0) { + return -EINVAL; + } + + if (mode == COMMIT_MODE_RETAIN) { + /* Commit state is fully derived from the registry watermarks at + * ACK time, so RETAIN registration has nothing to record per + * entry: return immediately, never block; the entry commits later + * via commit_on_ack (Req 2.1). */ + return 0; + } + + /* COMMIT_MODE_WAIT: blocking wait + admission cap land in task 2.2. */ + return -ENOSYS; } +/* ============================================================ + * ACK and membership events + * ============================================================ */ + void commit_on_ack(commit_engine_t* ce, int standby_idx, uint64_t acked_lsn) { - (void)ce; - (void)standby_idx; - (void)acked_lsn; - /* stub: never advances committed_lsn */ + if (ce == NULL || ce->p == NULL || standby_idx < 0 || + standby_idx >= MAX_STANDBYS) { + return; + } + + TRACE("commit: ack standby_idx=%d acked_lsn=%lu", standby_idx, acked_lsn); + + /* The authoritative watermark is read from the registry during + * evaluation (repl_on_ack stored it before this call); the args + * identify the event for tracing only. */ + pthread_mutex_lock(&ce->lock); + commit_advance_locked(ce); + pthread_mutex_unlock(&ce->lock); } void commit_set_writable_lsn(commit_engine_t* ce, uint64_t noop_lsn) { - (void)ce; - (void)noop_lsn; + if (ce == NULL) { + return; + } + pthread_mutex_lock(&ce->lock); + ce->writable_lsn = noop_lsn; + /* Gate change is part of the WAIT completion predicate: wake blockers + * so they re-evaluate (no successful completion until + * committed_lsn >= writable_lsn). */ + pthread_cond_broadcast(&ce->cond); + pthread_mutex_unlock(&ce->lock); } void commit_on_membership_change(commit_engine_t* ce) { - (void)ce; + if (ce == NULL || ce->p == NULL) { + return; + } + pthread_mutex_lock(&ce->lock); + /* Catch-up transitions / standby removal can newly satisfy a pending + * majority: re-evaluate (monotonic, so shrink never regresses) and + * wake all waiters to re-check their predicates. */ + commit_advance_locked(ce); + pthread_cond_broadcast(&ce->cond); + pthread_mutex_unlock(&ce->lock); }