diff --git a/src/controller/cluster.c b/src/controller/cluster.c index 931b48ee..98591703 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -609,7 +609,10 @@ int cluster_init(cluster_t* c) apiserver_init(c); - /* Persistence: derive paths, init, and recover from WAL/snapshot */ + /* Persistence: derive paths and init WAL/snapshot directories. + * Recovery is deferred until after cluster_build_resources() runs inside + * the HOOK_AFTER_EVENT below so that resmgr is fully populated before + * restore_worker() tries to register per-host resource values. */ { const char* workdir = c->myconf->workdir ? c->myconf->workdir : DEFAULT_WORKDIR; @@ -623,12 +626,6 @@ int cluster_init(cluster_t* c) ERROR("persist_init failed: %d", rc); return rc; } - rc = persist_recover(&c->persist); - if (rc != 0) { - ERROR("persist_recover failed: %d", rc); - persist_fini(&c->persist); - return rc; - } } /* HA initialisation */ @@ -684,6 +681,15 @@ int cluster_init(cluster_t* c) xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_CLUSTER, HOOK_AFTER_EVENT, HOOK_DETAIL_CLUSTER_INIT, c); + /* Recover WAL/snapshot now that resmgr, jobmgr, and workermgr are + * fully populated by cluster_build_resources() above. */ + rc = persist_recover(&c->persist); + if (rc != 0) { + ERROR("persist_recover failed: %d", rc); + persist_fini(&c->persist); + return rc; + } + return 0; } diff --git a/src/controller/join.c b/src/controller/join.c index 2db110ec..4eade071 100644 --- a/src/controller/join.c +++ b/src/controller/join.c @@ -61,10 +61,25 @@ static int handle_worker_join(cluster_t* c, endpoint_t* ep, HOOK_DETAIL_WORKER_JOIN, c, from->name, 0); idx = add_worker(c, ep, from->name, joinmsg->data.attrs); + /* A worker recovered from WAL already occupies the slot (status=UNAVAIL). + * Reconnecting after a controller restart will fail add_worker with -EEXIST. + * Locate the existing slot, update its endpoint, and clear UNAVAIL. */ + if (idx == -EEXIST) { + worker_t* existing = NULL; + if (key_find(c->workermgr, (void*)from->name, (void**)&existing) > 0 && + existing != NULL) { + lock(existing, RTTI_LOCK_WRITE); + existing->ep = ep; + existing->host.status &= ~XCE_WORKER_UNAVAIL; + idx = existing->idx; + unlock(existing); + } + } + reply.status = idx; reply.data_which = RequestReply_data_nodata; - /* Look up the newly added worker so the persist hook can serialize it */ + /* Look up the worker so the persist hook can serialize it */ worker_t* w = NULL; if (idx >= 0) vget(c->workermgr, idx, (void**)&w); diff --git a/src/controller/persist.c b/src/controller/persist.c index 81c36ff6..8909bb4e 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -1016,10 +1016,14 @@ int persist_job_hook_fun(int type, int situation, int detail, rc = -ENOMEM; break; } + /* persist_wal_sync_append returns 0 on successful local WAL write + * even when replication is degraded (no standbys → -ENODEV, timeout + * → -ETIMEDOUT). Only abort the loop on a real disk-write failure + * (non-zero return). Replication status is informational here. */ int sync_rc = 0; - persist_wal_sync_append(p, WAL_JOB_SUBMIT, buf, (uint32_t)sz, &sync_rc); + rc = persist_wal_sync_append(p, WAL_JOB_SUBMIT, buf, (uint32_t)sz, + &sync_rc); free(buf); - if (sync_rc != 0) rc = sync_rc; } return rc; } diff --git a/src/lib/xcectx.c b/src/lib/xcectx.c index 7aa3f267..0d0758a0 100644 --- a/src/lib/xcectx.c +++ b/src/lib/xcectx.c @@ -133,6 +133,29 @@ static int handle_msg(endpoint_t* ep) (void*)ep); } +/* Called by netio when the TCP connection drops (EPOLLHUP / EPOLLERR / read EOF). + * Unblocks any thread stuck in cbuf_get (e.g. xce_hello waiting for a reply) + * so that checking_connection can detect the failure and attempt to rejoin. */ +static int handle_disconnect(endpoint_t* ep) +{ + if (ep == NULL) + return 0; + + xce_ctx_t* ctx = (xce_ctx_t*)ep->owner; + if (ctx == NULL) + return 0; + + ctx->state = XCE_CTX_ERR; + + /* Inject an error into wbuf so any blocked cbuf_get returns immediately. + * cbuf_tryput returns -EAGAIN if the buffer already has an item — that is + * fine: the existing item will wake the waiter. */ + int err = -ECONNRESET; + cbuf_tryput(&ctx->wbuf, &err); + + return 0; +} + int xce_ctx_init(xce_ctx_t* ctx, char* name, char* key) { int rc; @@ -237,6 +260,7 @@ int xce_ctx_start(xce_ctx_t* ctx) endpoint_set_owner(ctx->ep, ctx); endpoint_set_msg_handler(ctx->ep, (msg_handler_t)handle_msg); + endpoint_set_err_handler(ctx->ep, (err_handler_t)handle_disconnect); rc = iomgr_monitor(ctx->ep); if (rc < 0) { ctx->state = XCE_CTX_ERR;