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

feat(ha-config-bootstrap): wire repl_wait_majority_ack into persist_wal_sync_append (task 7.6)

Browse source
This commit is contained in:
Rongsong Shen 2026-05-28 14:28:38 +00:00
parent 85c8565957
commit 684d3b6608
2 changed files with 21 additions and 8 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.kiro/specs/ha-config-bootstrap/tasks.md
View file

@ -181,7 +181,7 @@ Dependency order: startup validation (D) unblocks vtaskmgr before election (A);
- _Requirements: 5.6, 5.7_ - _Requirements: 5.6, 5.7_
- _Boundary: Persistence Layer (persist.c, persist.h)_ - _Boundary: Persistence Layer (persist.c, persist.h)_
- [ ] 7.6 Wire majority-commit into persist_wal_sync_append() - [x] 7.6 Wire majority-commit into persist_wal_sync_append()
- In `persist_wal_sync_append()`: after local write + fsync for `WAL_JOB_SUBMIT`, `WAL_JOB_START`, `WAL_JOB_FINISH`, `WAL_QUEUE_CHANGE` → call `repl_broadcast(c, OP_walAck, payload, len)` + `repl_wait_majority_ack(c, commit_timeout_ms, OP_walAck)` → on quorum advance `committed_lsn` - In `persist_wal_sync_append()`: after local write + fsync for `WAL_JOB_SUBMIT`, `WAL_JOB_START`, `WAL_JOB_FINISH`, `WAL_QUEUE_CHANGE` → call `repl_broadcast(c, OP_walAck, payload, len)` + `repl_wait_majority_ack(c, commit_timeout_ms, OP_walAck)` → on quorum advance `committed_lsn`
- `WAL_HOST_STATE`, `WAL_CHECKPOINT_REF` bypass to `persist_wal_async()` (unchanged) - `WAL_HOST_STATE`, `WAL_CHECKPOINT_REF` bypass to `persist_wal_async()` (unchanged)
- `make test` passes; integration test verifies committed WAL entries survive failover - `make test` passes; integration test verifies committed WAL entries survive failover

27
src/controller/persist.c
View file

@ -1005,21 +1005,34 @@ int persist_wal_sync_append(persist_t* p, uint32_t type,
return rc; return rc;
} }
/* For sync replication (WAL_JOB_SUBMIT): wait for at least one standby ACK. /* 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.
* We use current_lsn - 1 because persist_wal_async already incremented it. */ * We use current_lsn - 1 because persist_wal_async already incremented it. */
uint64_t written_lsn; uint64_t written_lsn;
pthread_mutex_lock(&p->lock); pthread_mutex_lock(&p->lock);
written_lsn = p->current_lsn - 1; written_lsn = p->current_lsn - 1;
pthread_mutex_unlock(&p->lock); pthread_mutex_unlock(&p->lock);
int sync_err = repl_wait_ack(&p->repl, written_lsn, /* No standbys registered → degraded mode immediately (no quorum to wait for) */
p->config.sync_timeout_ms); mylock_lock(&p->repl.lock);
if (sync_err < 0) int has_standbys = (p->repl.nstandbys > 0);
CTRL_METRIC_INC(ctrl_repl_degraded_count); mylock_unlock(&p->repl.lock);
int sync_err;
if (!has_standbys) {
sync_err = -ENODEV;
} else {
sync_err = repl_wait_majority_ack(&p->repl, 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) if (sync_rc_out)
*sync_rc_out = sync_err; *sync_rc_out = sync_err;
/* -ETIMEDOUT = degraded mode (local-only durability); -ENODEV = no standbys. /* sync_err < 0 = degraded mode (local-only durability); continue operating. */
* Both are acceptable — primary continues operating. */
return 0; return 0;
} }