diff --git a/.kiro/specs/ha-config-bootstrap/tasks.md b/.kiro/specs/ha-config-bootstrap/tasks.md index 5fb5d03d..12d21617 100644 --- a/.kiro/specs/ha-config-bootstrap/tasks.md +++ b/.kiro/specs/ha-config-bootstrap/tasks.md @@ -160,7 +160,7 @@ Dependency order: startup validation (D) unblocks vtaskmgr before election (A); - _Requirements: 5.1, 5.4_ - _Boundary: Replication Layer (repl.c, repl.h)_ -- [ ] 7.3 Implement repl_wait_majority_ack() +- [x] 7.3 Implement repl_wait_majority_ack() - `repl_wait_majority_ack(cluster_t *c, uint32_t timeout_ms, uint8_t ack_op)`: `cond_timedwait` until `acked_count + 1 >= (n_active + 1) / 2`; single-node shortcircuit; `ack_op` disambiguates WAL vs config ACKs; returns 0 on quorum, `-ETIMEDOUT` on timeout - `repl_on_ack()` extended to decrement `pending_ack_count` and signal condition - Log INFO on quorum (10.3); log WARN on timeout (10.3) diff --git a/src/controller/repl.c b/src/controller/repl.c index 13eb369e..41079e83 100644 --- a/src/controller/repl.c +++ b/src/controller/repl.c @@ -8,6 +8,7 @@ #include #include +#include "controller.capnp.h" /* OPCode_walAck, OPCode_upgradeConfigAck */ #include "repl.h" #include "mylock.h" #include "utils.h" @@ -95,6 +96,7 @@ int repl_add_standby(replicator_t *r, repl_ep_t ep, const char *host, uint32_t p strncpy(s->host, host, XCE_MAXNAME_LEN - 1); s->host[XCE_MAXNAME_LEN - 1] = '\0'; atomic_store(&s->acked_lsn, 0); + atomic_store(&s->acked_config_lsn, 0); atomic_store(&s->in_catch_up, 0); r->nstandbys++; @@ -165,13 +167,15 @@ void repl_on_ack(replicator_t *r, repl_ep_t ep, uint64_t acked_lsn) standby_conn_t *s = &r->standbys[i]; if (s->ep == ep) { atomic_store(&s->acked_lsn, acked_lsn); + if (s->pending_ack_count > 0) + s->pending_ack_count--; break; } } mylock_unlock(&r->lock); - /* Signal any waiter (repl_wait_ack) */ + /* Signal any waiter (repl_wait_ack / repl_wait_majority_ack) */ pthread_mutex_lock(&r->ack_lock); pthread_cond_signal(&r->ack_cond); pthread_mutex_unlock(&r->ack_lock); @@ -293,15 +297,120 @@ int repl_wait_ack(replicator_t *r, uint64_t target_lsn, uint32_t timeout_ms) } /* ============================================================ - * Majority-quorum ACK wait — stub until task 7.3 + * Config ACK handling — call from io_worker read callback ONLY (task 8.1) + * ============================================================ */ + +void repl_on_config_ack(replicator_t *r, repl_ep_t ep, uint64_t acked_config_lsn) +{ + if (r == NULL || ep == NULL) + return; + + mylock_lock(&r->lock); + for (int i = 0; i < MAX_STANDBYS; i++) { + standby_conn_t *s = &r->standbys[i]; + if (s->ep == ep) { + atomic_store_explicit(&s->acked_config_lsn, acked_config_lsn, + memory_order_release); + break; + } + } + mylock_unlock(&r->lock); + + pthread_mutex_lock(&r->ack_lock); + pthread_cond_signal(&r->ack_cond); + pthread_mutex_unlock(&r->ack_lock); +} + +/* ============================================================ + * Majority-quorum ACK wait (task 7.3) + * + * Quorum formula: quorum = (n_active + 2) / 2 [ceiling of (n_active+1)/2] + * n_active = count of active (non-catch-up) standbys. + * + * walAck: acked_total = 1 (self) + standbys with acked_lsn > 0. + * Primary counts itself — it persists WAL before broadcasting. + * configAck: acked_total = standbys with acked_config_lsn > 0 (no self-count). + * Config must be confirmed by standbys; primary does not self-count. + * + * Returns 0 on quorum, -ETIMEDOUT on timeout. * ============================================================ */ int repl_wait_majority_ack(replicator_t *r, uint32_t timeout_ms, uint8_t ack_op) { - (void)r; - (void)timeout_ms; - (void)ack_op; - return -ENOSYS; + if (r == NULL) + return -EINVAL; + + int is_wal_ack = (ack_op == (uint8_t)OPCode_walAck); + + struct timespec deadline; + clock_gettime(CLOCK_REALTIME, &deadline); + deadline.tv_sec += (long)(timeout_ms / 1000); + deadline.tv_nsec += (long)(timeout_ms % 1000) * 1000000L; + if (deadline.tv_nsec >= 1000000000L) { + deadline.tv_sec++; + deadline.tv_nsec -= 1000000000L; + } + + pthread_mutex_lock(&r->ack_lock); + int rc = -ETIMEDOUT; + + while (1) { + mylock_lock(&r->lock); + + int n_active = 0; + int acked_count = 0; + + for (int i = 0; i < MAX_STANDBYS; i++) { + standby_conn_t *s = &r->standbys[i]; + if (s->ep == NULL) + continue; + if (atomic_load_explicit(&s->in_catch_up, memory_order_acquire)) + continue; + n_active++; + if (is_wal_ack) { + if (atomic_load_explicit(&s->acked_lsn, memory_order_acquire) > 0) + acked_count++; + } else { + if (atomic_load_explicit(&s->acked_config_lsn, + memory_order_acquire) > 0) + acked_count++; + } + } + + mylock_unlock(&r->lock); + + /* Single-node shortcircuit: no active standbys → local write is quorum. + * Applies to both walAck (self=1, quorum=1) and configAck (Req 8.5). */ + if (n_active == 0) { + rc = 0; + break; + } + + int quorum = (n_active + 2) / 2; + int acked_total = is_wal_ack ? (1 + acked_count) : acked_count; + + if (acked_total >= quorum) { + rc = 0; + break; + } + + int wait_rc = pthread_cond_timedwait(&r->ack_cond, &r->ack_lock, + &deadline); + if (wait_rc == ETIMEDOUT) { + rc = -ETIMEDOUT; + break; + } + } + + pthread_mutex_unlock(&r->ack_lock); + + if (rc == 0) + INFO("repl_wait_majority_ack: quorum reached (ack_op=%u)", ack_op); + else + WARN("repl_wait_majority_ack: timeout waiting for majority quorum" + " (ack_op=%u, timeout_ms=%u)", ack_op, timeout_ms); + + return rc; } /* ============================================================ diff --git a/src/include/repl.h b/src/include/repl.h index 6d05fff7..3c76b9b0 100644 --- a/src/include/repl.h +++ b/src/include/repl.h @@ -20,9 +20,10 @@ typedef void* repl_ep_t; /* Per-standby connection slot */ typedef struct { - repl_ep_t ep; /* NULL = slot empty */ - _Atomic uint64_t acked_lsn; /* Last LSN ACKed by this standby */ - _Atomic int in_catch_up; /* Non-zero while catching up */ + repl_ep_t ep; /* NULL = slot empty */ + _Atomic uint64_t acked_lsn; /* Last WAL LSN ACKed (walAck) */ + _Atomic uint64_t acked_config_lsn; /* Last config LSN ACKed (upgradeConfigAck) */ + _Atomic int in_catch_up; /* Non-zero while catching up */ uint64_t join_applied_lsn; /* LSN standby reported on reconnect */ char host[XCE_MAXNAME_LEN]; uint32_t port; @@ -54,13 +55,16 @@ void repl_expire_disconnect_slots(replicator_t *r, time_t now_s, uint32_t snap_i int repl_broadcast(replicator_t *r, const wal_entry_hdr_t *hdr, const void *payload, uint32_t payload_len); -/* ACK handling — call from io_worker read callback ONLY */ +/* WAL ACK handling — call from io_worker read callback ONLY */ void repl_on_ack(replicator_t *r, repl_ep_t ep, uint64_t acked_lsn); +/* Config ACK handling — call from io_worker read callback ONLY (task 8.1) */ +void repl_on_config_ack(replicator_t *r, repl_ep_t ep, uint64_t acked_config_lsn); + /* Sync replication (stub in P2-1; wired in P2-3) */ int repl_wait_ack(replicator_t *r, uint64_t target_lsn, uint32_t timeout_ms); -/* Majority-quorum ACK wait — stub until task 7.3 */ +/* Majority-quorum ACK wait (task 7.3) */ int repl_wait_majority_ack(replicator_t *r, uint32_t timeout_ms, uint8_t ack_op); /* WAL trim guard */