feat(ha-config-bootstrap): implement repl_wait_majority_ack with quorum formula and ack_op discrimination (task 7.3)

.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)

src/controller/repl.c

@@ -8,6 +8,7 @@
 #include <errno.h>
 #include <string.h>
 
+#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;
+    if (r == NULL)
-    (void)timeout_ms;
+        return -EINVAL;
-    (void)ack_op;
+
-    return -ENOSYS;
+    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;
 }
 
 /* ============================================================

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 */
+    repl_ep_t          ep;                /* NULL = slot empty */
-    _Atomic uint64_t   acked_lsn;      /* Last LSN ACKed by this standby */
+    _Atomic uint64_t   acked_lsn;        /* Last WAL LSN ACKed (walAck) */
-    _Atomic int        in_catch_up;    /* Non-zero while catching up */
+    _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 */