feat(ha-config-bootstrap): extract cluster_open_listener, reorder startup before election (task 3.2)

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

@@ -59,7 +59,7 @@ Dependency order: startup validation (D) unblocks vtaskmgr before election (A);
   - `make test` fails
   - _Requirements: 1.1, 1.2, 1.3, 1.4_
 
-- [ ] 3.2 Extract cluster_open_listener() from cluster_init_phase2()
+- [x] 3.2 Extract cluster_open_listener() from cluster_init_phase2()
   - Move `bind()+listen()` + socket setup into new `cluster_open_listener(cluster_t *c)` in `cluster.c`
   - `cluster_init_phase2()` calls `cluster_open_listener()` for compatibility
   - `myapp.c` startup sequence calls `cluster_open_listener()` before election (after `cluster_init_phase1()`)

src/controller/cluster.c

@@ -641,6 +641,59 @@ int cluster_init_phase1(cluster_t* c)
     /* cbuf used by cluster_run() to signal restart/stop */
     cbuf_init(&(c->wbuf), 1, sizeof(int));
 
+    /* Sentinel: -1 means iomgr not yet initialised.
+     * cluster_init_phase2() uses this to skip re-init when myapp_run()
+     * has already called iomgr_init() before election. */
+    c->iomgr.epoll_fd = -1;
+
+    /* Sentinel: fd=-1 means the listener socket is not yet open.
+     * cluster_open_listener() uses this for its idempotency guard. */
+    c->ep.fd = -1;
+
+    return 0;
+}
+
+int cluster_open_listener(cluster_t* c)
+{
+    int rc;
+
+    if (c == NULL)
+        return -EINVAL;
+
+    /* Idempotent: if the listener socket is already open, do nothing. */
+    if (c->ep.fd >= 0)
+        return 0;
+
+    /* Determine listen port: prefer per-candidate port from --peers list,
+     * fall back to cluster_config->port if config is already loaded. */
+    uint32_t my_port = 0;
+    {
+        const xce_cluster_config_t* pcfg = &c->myconf->peers_cfg;
+        for (int j = 0; j < pcfg->candidates_count; j++) {
+            const char* key = pcfg->candidates[j].name[0]
+                                  ? pcfg->candidates[j].name
+                                  : pcfg->candidates[j].hostname;
+            if (strncmp(key, c->myconf->name, XCE_MAXNAME_LEN - 1) == 0) {
+                my_port = pcfg->candidates[j].port;
+                break;
+            }
+        }
+    }
+    if (my_port == 0) {
+        if (c->myconf->cluster_config != NULL) {
+            my_port = (uint32_t)c->myconf->cluster_config->port;
+        } else {
+            ERROR("cluster_open_listener: no port in --peers and cluster_config "
+                  "not yet loaded; specify port in --peers list");
+            return -EINVAL;
+        }
+    }
+
+    rc = endpoint_init(&(c->ep), AF_INET, SOCK_STREAM, (int)my_port, 5);
+    if (rc < 0)
+        return rc;
+
+    endpoint_set_peer(&(c->ep), c->myhost);
     return 0;
 }
 
@@ -705,39 +758,28 @@ int cluster_init_phase2(cluster_t* c)
                 (zarray_item_killer_t)resval_obj_killer);
     c->workermgr = makeInstance(&xkvmgr_type, &workermgr_params);
     c->qmgr      = makeInstance(&xkvmgr_type, &qmgr_params);
-    iomgr_init(&(c->iomgr));
 
-    iomgr_set_ssl_ctx(&(c->iomgr), ssl_ctx);
+    /* iomgr and vtaskmgr are initialised in cluster_open_listener() /
+     * myapp_run() before election.  Guard against re-initialisation when
+     * cluster_init() (legacy wrapper) calls cluster_init_phase2() directly. */
+    if (c->iomgr.epoll_fd < 0) {
+        iomgr_init(&(c->iomgr));
+        iomgr_set_ssl_ctx(&(c->iomgr), ssl_ctx);
+    } else {
+        /* ssl_ctx may have been set before phase1; always refresh it here. */
+        iomgr_set_ssl_ctx(&(c->iomgr), ssl_ctx);
+    }
 
     xstring_table_init(c->myconf->cluster_config->xstrings->nhash_slots);
     job_obj_init(5381, 5381);
 
     zarray_init(&(c->endpoints), 16, endpoint_killer);
 
-    /* Determine listen port: prefer per-candidate port from --peers list,
-     * fall back to cluster_config->port */
-    uint32_t my_port = 0;
-    {
-        const xce_cluster_config_t* pcfg = &c->myconf->peers_cfg;
-        for (int j = 0; j < pcfg->candidates_count; j++) {
-            const char* key = pcfg->candidates[j].name[0]
-                                  ? pcfg->candidates[j].name
-                                  : pcfg->candidates[j].hostname;
-            if (strncmp(key, c->myconf->name, XCE_MAXNAME_LEN - 1) == 0) {
-                my_port = pcfg->candidates[j].port;
-                break;
-            }
-        }
-    }
-    if (my_port == 0)
-        my_port = (uint32_t)c->myconf->cluster_config->port;
-
-    rc = endpoint_init(&(c->ep), AF_INET, SOCK_STREAM, (int)my_port, 5);
-    if (rc < 0) {
+    /* Open the listener socket (idempotent — skipped if already opened before
+     * election via cluster_open_listener()). */
+    rc = cluster_open_listener(c);
+    if (rc < 0)
         return rc;
-    }
-
-    endpoint_set_peer(&(c->ep), c->myhost);
 
     for (i = 0; i < c->ncontrollers; i++) {
         controller_init(&(c->controllers[i]), c);
@@ -746,8 +788,11 @@ int cluster_init_phase2(cluster_t* c)
     zarray_init(&(c->configs), 64, NULL /* FIXME */);
 
     /* Req 4.3: use --vcpus CLI value; fall back to built-in default (8).
-     * The cluster config 'vcpus' field has been removed (Req 4.4 deprecation). */
-    {
+     * The cluster config 'vcpus' field has been removed (Req 4.4 deprecation).
+     * vtaskmgr_init() is guarded by n_vcpus==0: myapp_run() calls it before
+     * election; the guard here covers the legacy cluster_init() path where
+     * it has not been called yet. */
+    if (c->vtaskmgr.n_vcpus == 0) {
         int nvcpus = (c->myconf->vcpus > 0) ? c->myconf->vcpus : 8;
         vtaskmgr_init(&(c->vtaskmgr), nvcpus);
     }

src/controller/include/cluster.h

@@ -23,6 +23,7 @@ typedef int (*collect_check_fn_t)(void*);
 
 extern int             check_controller(void*);
 extern int             cluster_init_phase1(cluster_t*);
+extern int             cluster_open_listener(cluster_t*);
 extern int             cluster_init_phase2(cluster_t*);
 extern int             cluster_init(cluster_t*);
 extern int             cluster_start(cluster_t*);

src/controller/myapp.c

@@ -48,6 +48,26 @@ int myapp_run(myconf_t* myconf)
     }
     c.ha.ssl_ctx = ssl_ctx;
 
+    /* Req 1.1: open the listener socket before any election probe so that
+     * concurrent-startup peers can reach this node's vote handler. */
+    rc = cluster_open_listener(&c);
+    if (rc < 0) {
+        ERROR("cluster_open_listener: %s", strerror(-rc));
+        cluster_fini(&c);
+        return rc;
+    }
+
+    /* Req 1.2: initialise the I/O manager and virtual task manager before
+     * election so that incoming vote requests are handled by the normal
+     * dispatch loop once it starts. */
+    iomgr_init(&(c.iomgr));
+    iomgr_set_ssl_ctx(&(c.iomgr), ssl_ctx);
+
+    {
+        int nvcpus = (myconf->vcpus > 0) ? myconf->vcpus : 8;
+        vtaskmgr_init(&(c.vtaskmgr), nvcpus);
+    }
+
     /* Election: determine role before loading config.
      * Capture result once — am_i_primary() probes peers and is expensive. */
     int is_primary = am_i_primary(&c);