feat(ha-config-bootstrap): implement committed_config_tag and config quorum commit (task 8.1)

.kiro/specs/ha-config-bootstrap/tasks.md

@@ -189,7 +189,7 @@ Dependency order: startup validation (D) unblocks vtaskmgr before election (A);
   - _Boundary: Persistence Layer (persist.c), Replication Layer (repl.c)_
 
 - [ ] 8. Change H: Majority-commit config replication (P)
-- [ ] 8.1 Implement OP_upgradeConfigAck handler and committed_config_tag
+- [x] 8.1 Implement OP_upgradeConfigAck handler and committed_config_tag
   - Add `committed_config_tag` field to cluster config state (ha_state_t via `last_known_committed_config_tag`); piggybacked on `CtrlHeartbeat` via `committedConfigTag`
   - Standby handler for `OP_upgradeConfig`: write config blob locally + reply `OP_upgradeConfigAck` with `ackedTag`
   - `push_config_to_standbys()` upgraded: write locally → `repl_broadcast(c, OP_upgradeConfig, blob, len)` → `repl_wait_majority_ack(c, commit_timeout_ms, OP_upgradeConfigAck)` → on quorum `atomic_store(&committed_config_tag, tag)` → return; on timeout return `-ETIMEDOUT`

src/controller/config_repl.c

@@ -27,7 +27,9 @@
 #include "byteorder.h"
 #include "clusterconfig.h"
 #include "cluster.h"
+#include "controller.capnp.h"  /* OPCode_upgradeConfigAck */
 #include "ha.h"
+#include "repl.h"              /* repl_wait_majority_ack, repl_on_config_ack */
 #include "request.h"
 #include "utils.h"
 #include "wobj.h"
@@ -1119,10 +1121,24 @@ int push_config_to_standbys(cluster_t *c, const uint8_t *bytes, size_t blen,
     int acked = wait_all_with_timeout(threads, ctxs, n,
                                        CONF_PUSH_TIMEOUT_MS, &push_done);
 
-    INFO("Config pushed to %d/%d standbys", acked, n);
-
     free(ctxs);
     free(threads);
+
+    /* Req 8.1–8.3: wait for majority-quorum OP_upgradeConfigAck.
+     * On quorum: advance committed_config_tag atomically (Req 8.5). */
+    int rc = repl_wait_majority_ack(&c->persist.repl, CONF_PUSH_TIMEOUT_MS,
+                                    (uint8_t)OPCode_upgradeConfigAck);
+    if (rc == 0) {
+        atomic_store_explicit(&c->ha.committed_config_tag, new_tag,
+                              memory_order_release);
+        INFO("Config committed on majority (tag=%016" PRIx64
+             ", pushed=%d/%d standbys)", new_tag, acked, n);
+    } else {
+        WARN("Config push: timeout waiting for majority ACK"
+             " (tag=%016" PRIx64 ", pushed=%d/%d standbys)",
+             new_tag, acked, n);
+    }
+
     return acked;
 }
 

src/controller/ha.h

@@ -156,6 +156,7 @@ typedef struct {
     _Atomic uint64_t last_known_committed_lsn;        /* Req 5: from primary heartbeats */
     sem_t            heartbeat_seen;                  /* Req 1: posted by heartbeat handler during CTRL_INIT */
     uint64_t         last_known_committed_config_tag; /* Req 8: from primary heartbeats */
+    _Atomic uint64_t committed_config_tag;            /* Req 8.1: set on quorum ACK */
 } ha_state_t;
 
 /* ── Public API ────────────────────────────────────────────── */