fix: correct controller restart recovery — jobs, workers, and resource ordering

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

@@ -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;
 }
 

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);

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;
     }

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;