feat(ha-commit-engine): repl-side wiring — ACK feed, membership hooks, adapter (task 5.2)
replicator_t gains ce pointer (set-once-before-traffic contract). repl_on_ack feeds commit_on_ack after releasing r->lock (iomgr-safe); four membership mutators feed commit_on_membership_change post-unlock, gated on actual mutation. walAck branch delegates to commit_submit(WAIT) when ce wired AND entry_lsn != 0 (LSN-0 interim guard until 5.3); legacy counting kept for ce==NULL, marked DEPRECATED. commit.c linked into production target (inert until 5.3 wiring). Tests 30-31 + all repl_majority walAck cases green; 33/34 remain RED for 5.3; baseline unchanged.
This commit is contained in:
parent
d1353360cc
commit
6600fd5d1f
6 changed files with 104 additions and 15 deletions
|
|
@ -1,5 +1,6 @@
|
|||
controller_src = [
|
||||
'cluster.c',
|
||||
'commit.c',
|
||||
'config_repl.c',
|
||||
'dataex.c',
|
||||
'ha.c',
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@
|
|||
|
||||
#include "controller.capnp.h" /* OPCode_walAck, OPCode_upgradeConfigAck */
|
||||
#include "repl.h"
|
||||
#include "commit.h" /* commit engine feed + adapter (ha-commit-engine 5.2) */
|
||||
#include "mylock.h"
|
||||
#include "utils.h"
|
||||
#include "request.h" /* xce_wal_batch_req_raw */
|
||||
|
|
@ -120,11 +121,21 @@ int repl_add_standby(replicator_t *r, repl_ep_t ep, const char *host, uint32_t p
|
|||
r->nstandbys++;
|
||||
|
||||
mylock_unlock(&r->lock);
|
||||
|
||||
/* Quorum membership changed: re-evaluate pending commits. Called AFTER
|
||||
* releasing r->lock — engine lock order is ce->lock then r->lock
|
||||
* (ha-commit-engine task 5.2, Req 4.2). The idempotent early-return
|
||||
* above mutates nothing, so it deliberately skips this hook. */
|
||||
if (r->ce != NULL) {
|
||||
commit_on_membership_change(r->ce);
|
||||
}
|
||||
return idx;
|
||||
}
|
||||
|
||||
void repl_remove_standby(replicator_t *r, repl_ep_t ep)
|
||||
{
|
||||
int removed = 0;
|
||||
|
||||
if (r == NULL || ep == NULL) {
|
||||
return;
|
||||
}
|
||||
|
|
@ -138,11 +149,18 @@ void repl_remove_standby(replicator_t *r, repl_ep_t ep)
|
|||
s->last_disconnect_time = time(NULL);
|
||||
s->ep = NULL;
|
||||
r->nstandbys--;
|
||||
removed = 1;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
mylock_unlock(&r->lock);
|
||||
|
||||
/* Membership shrink: wake pending waiters to re-check their predicates
|
||||
* (after r->lock release — engine lock order; ha-commit-engine Req 4.2). */
|
||||
if (removed && r->ce != NULL) {
|
||||
commit_on_membership_change(r->ce);
|
||||
}
|
||||
}
|
||||
|
||||
void repl_expire_disconnect_slots(replicator_t *r, time_t now_s, uint32_t snap_interval_s)
|
||||
|
|
@ -175,6 +193,8 @@ void repl_expire_disconnect_slots(replicator_t *r, time_t now_s, uint32_t snap_i
|
|||
|
||||
void repl_on_ack(replicator_t *r, repl_ep_t ep, uint64_t acked_lsn)
|
||||
{
|
||||
int slot_idx = -1;
|
||||
|
||||
if (r == NULL || ep == NULL) {
|
||||
return;
|
||||
}
|
||||
|
|
@ -187,12 +207,21 @@ void repl_on_ack(replicator_t *r, repl_ep_t ep, uint64_t acked_lsn)
|
|||
atomic_store(&s->acked_lsn, acked_lsn);
|
||||
if (s->pending_ack_count > 0)
|
||||
s->pending_ack_count--;
|
||||
slot_idx = i;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
mylock_unlock(&r->lock);
|
||||
|
||||
/* Feed the commit engine AFTER the acked_lsn store and AFTER releasing
|
||||
* r->lock — commit_on_ack takes ce->lock then r->lock (majority scan),
|
||||
* so calling it under r->lock would invert the engine's lock order.
|
||||
* Never blocks (iomgr-safe; ha-commit-engine task 5.2, design §ACK Feed). */
|
||||
if (r->ce != NULL && slot_idx >= 0) {
|
||||
commit_on_ack(r->ce, slot_idx, acked_lsn);
|
||||
}
|
||||
|
||||
/* Signal any waiter (repl_wait_ack / repl_wait_majority_ack) */
|
||||
pthread_mutex_lock(&r->ack_lock);
|
||||
pthread_cond_signal(&r->ack_cond);
|
||||
|
|
@ -386,17 +415,31 @@ void repl_on_config_ack(replicator_t *r, repl_ep_t ep, uint64_t acked_config_lsn
|
|||
int repl_wait_majority_ack(replicator_t *r, uint64_t entry_lsn,
|
||||
uint32_t timeout_ms, uint8_t ack_op)
|
||||
{
|
||||
/* entry_lsn is accepted but unused by this legacy counting body
|
||||
* (ha-commit-engine task 5.1 signature change): task 5.2 delegates the
|
||||
* walAck branch to commit_submit(ce, entry_lsn, COMMIT_MODE_WAIT, ...)
|
||||
* when the engine is wired. Behavior here stays byte-identical. */
|
||||
(void)entry_lsn;
|
||||
|
||||
if (r == NULL)
|
||||
return -EINVAL;
|
||||
|
||||
int is_wal_ack = (ack_op == (uint8_t)OPCode_walAck);
|
||||
|
||||
/* Engine adapter (ha-commit-engine task 5.2, design §Job-Submit Adapter):
|
||||
* the walAck barrier is per-LSN commit evaluation owned by the commit
|
||||
* engine when wired. Caller holds no repl locks here, so the engine's
|
||||
* ce->lock -> r->lock order is preserved.
|
||||
*
|
||||
* entry_lsn == 0 falls through to legacy counting: persist.c's
|
||||
* persist_wal_sync_append still passes 0 until task 5.3 forwards
|
||||
* written_lsn (and commit_submit rejects lsn==0 with -EINVAL). Note the
|
||||
* first-ever WAL entry IS LSN 0 — structurally un-commitable under the
|
||||
* current numbering; task 5.3 documents the resolution, and the
|
||||
* promotion no-op entry (ha-single-leader-role) eliminates it in
|
||||
* practice. configAck callers pass entry_lsn=0 and never delegate. */
|
||||
if (is_wal_ack && r->ce != NULL && entry_lsn != 0) {
|
||||
return commit_submit(r->ce, entry_lsn, COMMIT_MODE_WAIT, timeout_ms);
|
||||
}
|
||||
|
||||
/* DEPRECATED legacy counting below (connectivity-based, not per-LSN):
|
||||
* survives only for engine-less (ce == NULL) unit-test contexts and the
|
||||
* configAck branch; retired by ha-commit-barrier-rollout. */
|
||||
|
||||
struct timespec deadline;
|
||||
clock_gettime(CLOCK_REALTIME, &deadline);
|
||||
deadline.tv_sec += (long)(timeout_ms / 1000);
|
||||
|
|
@ -513,30 +556,50 @@ uint64_t repl_trim_lsn(replicator_t *r, uint64_t checkpoint_lsn,
|
|||
|
||||
void repl_start_catchup(replicator_t *r, repl_ep_t ep, uint64_t join_applied_lsn)
|
||||
{
|
||||
int changed = 0;
|
||||
|
||||
mylock_lock(&r->lock);
|
||||
for (int i = 0; i < MAX_STANDBYS; i++) {
|
||||
if (r->standbys[i].ep == ep) {
|
||||
r->standbys[i].join_applied_lsn = join_applied_lsn;
|
||||
atomic_store_explicit(&r->standbys[i].in_catch_up, 1,
|
||||
memory_order_release);
|
||||
changed = 1;
|
||||
break;
|
||||
}
|
||||
}
|
||||
mylock_unlock(&r->lock);
|
||||
|
||||
/* Catch-up entry excludes the standby from the evaluation count:
|
||||
* re-evaluate pending commits (after r->lock release — engine lock
|
||||
* order; ha-commit-engine task 5.2, Req 4.2). */
|
||||
if (changed && r->ce != NULL) {
|
||||
commit_on_membership_change(r->ce);
|
||||
}
|
||||
}
|
||||
|
||||
void repl_complete_catchup(replicator_t *r, repl_ep_t ep)
|
||||
{
|
||||
int changed = 0;
|
||||
|
||||
mylock_lock(&r->lock);
|
||||
for (int i = 0; i < MAX_STANDBYS; i++) {
|
||||
if (r->standbys[i].ep == ep) {
|
||||
atomic_store_explicit(&r->standbys[i].in_catch_up, 0,
|
||||
memory_order_release);
|
||||
changed = 1;
|
||||
break;
|
||||
}
|
||||
}
|
||||
mylock_unlock(&r->lock);
|
||||
|
||||
/* Standby rejoins the evaluation count: its watermark may newly
|
||||
* satisfy a pending majority — re-evaluate (after r->lock release —
|
||||
* engine lock order; ha-commit-engine task 5.2, Req 4.2). */
|
||||
if (changed && r->ce != NULL) {
|
||||
commit_on_membership_change(r->ce);
|
||||
}
|
||||
|
||||
/* Broadcast so any blocked repl_wait_ack() re-checks the ACK condition */
|
||||
pthread_mutex_lock(&r->ack_lock);
|
||||
pthread_cond_broadcast(&r->ack_cond);
|
||||
|
|
@ -563,16 +626,17 @@ int repl_needs_snapshot(replicator_t *r, uint64_t standby_lsn,
|
|||
* (ha.c) for write admission. Takes ONLY r->lock — never the commit-engine
|
||||
* mutex (lock-ordering rule: gate path holds no locks when calling in).
|
||||
* (ha-commit-engine Req 3.1, 3.2, 3.3) */
|
||||
/* repl_set_commit_engine — INERT stub (ha-commit-engine task 5.1).
|
||||
* Stores nothing, so the engine is never wired and the task-5.1 RED tests
|
||||
* fail on behavior (ACK and membership events never reach the engine),
|
||||
* not on linking. Task 5.2 adds the replicator_t.ce field, stores the
|
||||
* pointer here, and feeds commit_on_ack / commit_on_membership_change
|
||||
* from repl_on_ack and the membership mutators. */
|
||||
/* repl_set_commit_engine — wire the commit engine into the replicator
|
||||
* (ha-commit-engine task 5.2). Once set, repl_on_ack feeds commit_on_ack,
|
||||
* the membership mutators feed commit_on_membership_change, and the walAck
|
||||
* branch of repl_wait_majority_ack delegates to commit_submit. Called once
|
||||
* at cluster init before replication traffic (task 5.3); NULL un-wires. */
|
||||
void repl_set_commit_engine(replicator_t *r, struct commit_engine *ce)
|
||||
{
|
||||
(void)r;
|
||||
(void)ce;
|
||||
if (r == NULL) {
|
||||
return;
|
||||
}
|
||||
r->ce = ce;
|
||||
}
|
||||
|
||||
int repl_active_standby_count(replicator_t *r)
|
||||
|
|
|
|||
|
|
@ -44,6 +44,12 @@ typedef struct replicator_s {
|
|||
mylock_t lock; /* Protects standbys[], nstandbys */
|
||||
pthread_cond_t ack_cond; /* Signaled when any ACK arrives */
|
||||
pthread_mutex_t ack_lock; /* Guards ack_cond */
|
||||
/* Commit engine feed (ha-commit-engine task 5.2). NULL = not wired
|
||||
* (legacy counting contexts). Set once via repl_set_commit_engine at
|
||||
* cluster init, before replication traffic — read without r->lock.
|
||||
* Lock ordering: engine paths take ce->lock then r->lock, so repl.c
|
||||
* must NEVER call into the engine while holding r->lock. */
|
||||
struct commit_engine *ce;
|
||||
} replicator_t;
|
||||
|
||||
/* Lifecycle */
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue