wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity

feat(ha-term-safety): on_commit_advanced barrier trigger + commit-gate ownership (task 2.4)

Browse source
This commit is contained in:
Rongsong Shen 2026-06-22 07:33:04 +00:00
parent 2c5d298cb5
commit 3470bdfb71
6 changed files with 363 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

55
src/controller/commit.c
View file

@ -94,9 +94,30 @@ static uint64_t majority_candidate_locked(commit_engine_t* ce)
int n = 0;
int need, k, i;
/* ha-term-safety task 2.4: the raft library's current-term-gated commit
* index is the COMMIT SOURCE. Once fed, it is an ADDITIONAL ceiling on top
* of the durable cap: committed_lsn = min(durable-majority, raft_commit_index),
* so the barrier never advances committed over an entry the library has not
* admitted under its Figure-8 current-term rule (replacing the term-blind
* safe-commit). Folding it into `cap` makes BOTH the standalone (n_peers<=1)
* and the multi-peer paths honor the source uniformly.
*
* FEATURE GATE: raft_commit_index == 0 means the library commit source is
* NOT yet active (no on_commit_advanced has fired — e.g. before tasks 3.x
* drive raft_advance_commit, or in a node that has committed nothing). In
* that state the ceiling is INERT and the barrier keeps its autonomous
* durable-majority behavior; the source only takes over once the library
* reports its first strict commit (commit_index >= 1, which is the lowest
* real LSN). This is the stub→real transition for this callback. */
if (ce->raft_commit_index != 0 && ce->raft_commit_index < cap) {
cap = ce->raft_commit_index;
}
/* Standalone / single-member: self constitutes the majority (Req 6.3) —
* but self still only counts up to what is durable locally, so a
* single-member commit also cannot advance over a not-yet-synced entry. */
* but self still only counts up to what is durable locally AND what the
* library has committed (cap already folds in raft_commit_index), so a
* single-member commit also cannot advance over a not-yet-synced or
* not-yet-library-committed entry. */
if (ce->n_peers <= 1) {
return cap;
}
@ -249,6 +270,7 @@ int commit_engine_init(commit_engine_t* ce, persist_t* p, int n_peers, int n_vcp
ce->n_vcpus = n_vcpus;
ce->inflight_waits = 0;
ce->writable_lsn = 0; /* no own-term gate until set by promotion */
ce->raft_commit_index = 0; /* no library commit source until fed (task 2.4) */
memset(ce->scratch, 0, sizeof(ce->scratch));
/* Register the durable-advance wake so the flusher can re-trigger commit
@ -394,6 +416,35 @@ void commit_set_writable_lsn(commit_engine_t* ce, uint64_t noop_lsn)
pthread_mutex_unlock(&ce->lock);
}
/* ha-term-safety task 2.4: feed the raft library's current-term-gated commit
* index as the commit SOURCE (R4.1, R4.5). Monotonic max() — the library only
* fires on_commit_advanced on a strict advance, but guard against a stale/lower
* value so the ceiling never regresses. Re-evaluate the barrier under the new
* ceiling and wake waiters: the durable cap may already cover the newly-admitted
* LSNs, so the commit can advance immediately without waiting for the next ACK.
*
* LOCK ORDERING: takes only ce->lock (commit_advance_locked re-reads the durable
* watermark + registry watermarks via the existing path). The on_commit_advanced
* callback runs on the iomgr ACK path that drove raft_advance_commit; that path
* does not hold p->lock here, preserving the p->lock → ce->lock direction. */
void commit_set_raft_commit_index(commit_engine_t* ce, uint64_t commit_index)
{
if (ce == NULL) {
return;
}
pthread_mutex_lock(&ce->lock);
if (commit_index > ce->raft_commit_index) {
ce->raft_commit_index = commit_index; /* monotonic ceiling */
}
/* The ceiling moved (or was confirmed): re-derive the majority under it and
* advance committed_lsn if the durable majority already covers the newly
* admitted prefix. commit_advance_locked broadcasts on any advance; broadcast
* unconditionally so a waiter re-checks even on a no-advance ceiling bump. */
commit_advance_locked(ce);
pthread_cond_broadcast(&ce->cond);
pthread_mutex_unlock(&ce->lock);
}
void commit_on_membership_change(commit_engine_t* ce)
{
if (ce == NULL || ce->p == NULL) {