diff --git a/.kiro/specs/ha-config-bootstrap/tasks.md b/.kiro/specs/ha-config-bootstrap/tasks.md index 84e5a5e..a505782 100644 --- a/.kiro/specs/ha-config-bootstrap/tasks.md +++ b/.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()`) diff --git a/src/controller/cluster.c b/src/controller/cluster.c index 0d0b11f..f21291a 100644 --- a/src/controller/cluster.c +++ b/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); } diff --git a/src/controller/include/cluster.h b/src/controller/include/cluster.h index 25b8c75..c42cbb7 100644 --- a/src/controller/include/cluster.h +++ b/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*); diff --git a/src/controller/myapp.c b/src/controller/myapp.c index e963e6d..866e020 100644 --- a/src/controller/myapp.c +++ b/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);