From 6429ef0bff3c08683e8a840c9c7fd760ebb00ce7 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sat, 9 May 2026 03:02:39 +0000 Subject: [PATCH] =?UTF-8?q?fix:=20correct=20controller=20restart=20recover?= =?UTF-8?q?y=20=E2=80=94=20jobs,=20workers,=20and=20resource=20ordering?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four bugs fixed: 1. persist.c: array job WAL loop aborted after index [0] when replication returned -ENODEV (no standbys). Loop guard now uses the return value of persist_wal_sync_append() which is 0 on local-disk success regardless of replication status. 2. join.c: WAL-recovered worker slot (status=UNAVAIL) caused add_worker() to return -EEXIST on reconnect. New EEXIST path in handle_worker_join() looks up the existing slot by name, updates its endpoint, and clears UNAVAIL so the normal hook/reply/on_worker_join flow proceeds. 3. xcectx.c: cbuf_get() in xce_hello() blocked indefinitely when the TCP connection dropped because no error handler was registered. Added handle_disconnect() which sets ctx->state=XCE_CTX_ERR and calls cbuf_tryput(-ECONNRESET) to unblock the waiter, allowing checking_connection() to detect the failure and rejoin. 4. cluster.c: persist_recover() was called before cluster_build_resources() populated resmgr, causing restore_worker() → add_hostattrs_resvals() to fail with EEXIST noise. Moved persist_recover() to after HOOK_AFTER_EVENT so resmgr is fully populated before WAL replay. --- src/controller/cluster.c | 20 +++++++++++++------- src/controller/join.c | 17 ++++++++++++++++- src/controller/persist.c | 8 ++++++-- src/lib/xcectx.c | 24 ++++++++++++++++++++++++ 4 files changed, 59 insertions(+), 10 deletions(-) diff --git a/src/controller/cluster.c b/src/controller/cluster.c index 931b48e..9859170 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 2db110e..4eade07 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 81c36ff..8909bb4 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 7aa3f26..0d0758a 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;