feat: P0-2a C-side plumbing for host control

Add the full call chain for host control operations from the Go API
server through the C library to the controller:

- schema/controller.capnp: HostControl struct + hostctrlmsg @21
  in Request.body union
- include/typemap.h: xce_hostctrl_t* hostctrlmsg in body union
  (placed after @20 clrpreasonmsg, matching capnp ordinal)
- include/client.h, xceapi.h, request.h: declarations for
  xcecli_hostctrl, xce_hostctrl, xce_hostctrl_req
- lib/client.c: xcecli_hostctrl (validate + delegate to xce_hostctrl)
- lib/xceapi.c: xce_hostctrl (validate ctx state + send + wait reply)
- lib/request.c: xce_hostctrl_req (validate hostname/empty/nil,
  accept only HOST_OPEN/HOST_CLOSE, reject HOST_TAG/UNTAG)
- lib/helper.c: Request_body_hostctrlmsg cases in
  xce_request_body() and xce_request_msg_name()
- controller/cluster.c: OPCode_hostcontrol dispatch
- controller/include/cluster.h: do_hostcontrol_op declaration
- controller/worker.c: do_hostcontrol_op + fix pre-existing
  goto-clean bug in worker_ctrl that leaked write lock and
  reference on HOST_OPEN/HOST_CLOSE no-op paths

TDD: 8 xce_hostctrl_req validation tests (null ep, null ctrl,
null/empty hostname, HOST_DO_NONE no-op, HOST_TAG/UNTAG reject,
negative action reject, HOST_OPEN/CLOSE accept via xce_request).

116/116 xcelib tests pass. CGO_ENABLED=0 Go backend tests pass.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

src/controller/cluster.c

@@ -189,6 +189,9 @@ static int myreq_handler(vtask_t* task)
     case OPCode_queuecontrol:
         rc = do_queuecontrol_op(c, ep, &request);
         break;
+    case OPCode_hostcontrol:
+        rc = do_hostcontrol_op(c, ep, &request);
+        break;
     case OPCode_runtasks:
         rc = do_runtasks(c, ep, &request);
         break;

src/controller/include/cluster.h

@@ -108,6 +108,7 @@ extern int update_resval_list(cluster_t*, worker_t*, xce_resval_list_t*);
 extern int do_jobcontrol_op(cluster_t*, endpoint_t*, xce_request_t*);
 extern int job_ctrl(cluster_t*, xce_jobctrl_t*);
 extern int do_queuecontrol_op(cluster_t*, endpoint_t*, xce_request_t*);
+extern int do_hostcontrol_op(cluster_t*, endpoint_t*, xce_request_t*);
 extern int do_runtasks(cluster_t*, endpoint_t*, xce_request_t*);
 extern int on_task_start(cluster_t*, xce_task_pidinfo_t*);
 extern int on_task_fail(cluster_t*, xce_task_failinfo_t*);

src/controller/worker.c

@@ -626,10 +626,23 @@ int worker_ctrl(cluster_t* c, xce_hostctrl_t* ctrl)
                       HOOK_AFTER_EVENT, HOOK_DETAIL_WORKER_UPDATE, c, w);
     }
 
-    unlock(w);
-
-    dereference(w);
-
 clean:
+    unlock(w);
+    dereference(w);
+    return rc;
+}
+
+int do_hostcontrol_op(cluster_t* c, endpoint_t* ep, xce_request_t* request)
+{
+    int         rc = 0;
+    xce_reply_t reply;
+
+    init_reply_by_req(&reply, request);
+
+    rc               = worker_ctrl(c, request->body.hostctrlmsg);
+    reply.status     = rc;
+    reply.data_which = RequestReply_data_nodata;
+
+    xce_reply(ep, c->myhost, &reply);
     return rc;
 }

src/include/client.h

@@ -54,6 +54,7 @@ extern int  xcecli_notify(xcecli_ctx_t, xce_notify_data_t*);
 extern int  xcecli_jobctrl(xcecli_ctx_t, xce_jobctrl_t*);
 extern int  xcecli_notify_jobchkpt(xcecli_ctx_t, uint64_t, char*);
 extern int  xcecli_queuectrl(xcecli_ctx_t, xce_queuectrl_t*);
+extern int  xcecli_hostctrl(xcecli_ctx_t, xce_hostctrl_t*);
 extern int  xcecli_runtask(xcecli_ctx_t, xce_taskrun_t*);
 extern int  xcecli_notify_taskstart(xcecli_ctx_t, xce_task_pidinfo_t*);
 extern int  xcecli_notify_taskfail(xcecli_ctx_t, xce_task_failinfo_t*);

src/include/request.h

@@ -54,6 +54,7 @@ extern int extract_data(dataset_t*, xce_data_t*);
 extern int xce_update_req(endpoint_t*, xce_update_data_t*);
 extern int xce_jobctrl_req(endpoint_t*, xce_jobctrl_t*);
 extern int xce_queuectrl_req(endpoint_t*, xce_queuectrl_t*);
+extern int xce_hostctrl_req(endpoint_t*, xce_hostctrl_t*);
 extern int xce_runtask_req(endpoint_t*, xce_taskrun_t*);
 extern int xce_set_pend_reason_req(endpoint_t*, xce_preason_set_t*);
 extern int xce_clear_reason_req(endpoint_t*, xce_clear_reason_t*);

src/include/typemap.h

@@ -196,6 +196,7 @@ typedef struct {
         xce_taskrun_t*      runtaskmsg;
         xce_preason_set_t*  preasonmsg;
         xce_clear_reason_t* clrpreasonmsg;
+        xce_hostctrl_t*     hostctrlmsg;
     } body;
 } xce_request_t;
 

src/include/xceapi.h

@@ -26,6 +26,7 @@ extern int xce_notify(xce_ctx_t*, xce_notify_data_t*);
 extern int xce_update_resvals(xce_ctx_t*, int, xce_resval_t**);
 extern int xce_jobctrl(xce_ctx_t*, xce_jobctrl_t*);
 extern int xce_queuectrl(xce_ctx_t*, xce_queuectrl_t*);
+extern int xce_hostctrl(xce_ctx_t*, xce_hostctrl_t*);
 extern int xce_runtask(xce_ctx_t*, xce_taskrun_t*);
 extern int xce_update_jusage(xce_ctx_t*, uint64_t, int, int*, int, int*, int,
                              xce_rusage_t**);

src/lib/client.c

@@ -516,6 +516,19 @@ int xcecli_queuectrl(xcecli_ctx_t c, xce_queuectrl_t* ctrlinfo)
     return rc;
 }
 
+int xcecli_hostctrl(xcecli_ctx_t c, xce_hostctrl_t* ctrlinfo)
+{
+    api_ctx_t* ctx = (api_ctx_t*)c;
+    int        rc;
+
+    if ((ctx == NULL) || (ctrlinfo == NULL)) {
+        return -EINVAL;
+    }
+
+    rc = xce_hostctrl(&(ctx->ctx), ctrlinfo);
+    return rc;
+}
+
 int xcecli_runtask(xcecli_ctx_t c, xce_taskrun_t* info)
 {
     api_ctx_t* ctx = (api_ctx_t*)c;

src/lib/helper.c

@@ -39,6 +39,8 @@ void* xce_request_body(xce_request_t* req)
         return req->body.jobctrlmsg;
     case Request_body_queuectrlmsg:
         return req->body.queuectrlmsg;
+    case Request_body_hostctrlmsg:
+        return req->body.hostctrlmsg;
     case Request_body_runtaskmsg:
         return req->body.runtaskmsg;
     case Request_body_preasonmsg:
@@ -85,6 +87,8 @@ char* xce_request_msg_name(xce_request_t* req)
         return "jobctrlmsg";
     case Request_body_queuectrlmsg:
         return "queuectrlmsg";
+    case Request_body_hostctrlmsg:
+        return "hostctrlmsg";
     case Request_body_runtaskmsg:
         return "runtaskmsg";
     case Request_body_preasonmsg:

src/lib/meson.build

@@ -146,6 +146,7 @@ libxce_test_src = ['testsuite/test.c',
                    'testsuite/test_jobj.c',
                    'testsuite/test_wobj.c',
                    'testsuite/test_qobj.c',
+                   'testsuite/test_request.c',
                   ]
 
 libxce_test_deps = [libxce_deps, librtti_dep, unity_dep, libxce_dep]

src/lib/request.c

@@ -844,6 +844,37 @@ int xce_queuectrl_req(endpoint_t* ep, xce_queuectrl_t* ctrlinfo)
     return xce_request(ep, &request);
 }
 
+int xce_hostctrl_req(endpoint_t* ep, xce_hostctrl_t* ctrlinfo)
+{
+    /* clang-format off */
+    xce_request_t request = {
+        .opcode = OPCode_hostcontrol,
+        .body_which = Request_body_hostctrlmsg,
+        .body = {
+            .hostctrlmsg = ctrlinfo
+        }
+    };
+    /* clang-format on */
+
+    if ((ep == NULL) || (ctrlinfo == NULL)) {
+        return -EINVAL;
+    }
+
+    if ((ctrlinfo->hostname == NULL) || (strlen(ctrlinfo->hostname) == 0)) {
+        return -EINVAL;
+    }
+
+    if (ctrlinfo->action == HOST_DO_NONE) {
+        return 0;
+    }
+
+    if (ctrlinfo->action != HOST_OPEN && ctrlinfo->action != HOST_CLOSE) {
+        return -EINVAL;
+    }
+
+    return xce_request(ep, &request);
+}
+
 int xce_runtask_req(endpoint_t* ep, xce_taskrun_t* runinfo)
 {
     /* clang-format off */

src/lib/testsuite/test.c

@@ -27,6 +27,7 @@ extern void test_kv(void);
 extern void test_jobobj(void);
 extern void test_qobj(void);
 extern void test_wobj(void);
+extern void test_request(void);
 
 void setUp(void) { return; }
 
@@ -123,6 +124,7 @@ int main(int argc, char* argv[])
     TEST(test_jobobj);
     TEST(test_wobj);
     TEST(test_qobj);
+    TEST(test_request);
 
     return UNITY_END();
 }

src/lib/testsuite/test_request.c

@@ -0,0 +1,79 @@
+#include <errno.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unity/unity.h>
+#include "request.h"
+#include "xce/types.h"
+
+/* Tests for xce_hostctrl_req() validation paths.
+ * We test only early-return paths (before xce_request() is called)
+ * since a live endpoint is not available in unit tests. */
+
+static void test_hostctrl_req_null_ep(void)
+{
+    xce_hostctrl_t ctrl = { .action = HOST_OPEN, .hostname = "host1" };
+    int rc = xce_hostctrl_req(NULL, &ctrl);
+    TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
+}
+
+static void test_hostctrl_req_null_ctrl(void)
+{
+    /* non-NULL dummy ep is enough; the NULL ctrl check fires first */
+    int rc = xce_hostctrl_req((endpoint_t*)0x1, NULL);
+    TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
+}
+
+static void test_hostctrl_req_null_hostname(void)
+{
+    xce_hostctrl_t ctrl = { .action = HOST_OPEN, .hostname = NULL };
+    int rc = xce_hostctrl_req((endpoint_t*)0x1, &ctrl);
+    TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
+}
+
+static void test_hostctrl_req_empty_hostname(void)
+{
+    xce_hostctrl_t ctrl = { .action = HOST_OPEN, .hostname = "" };
+    int rc = xce_hostctrl_req((endpoint_t*)0x1, &ctrl);
+    TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
+}
+
+static void test_hostctrl_req_noop(void)
+{
+    xce_hostctrl_t ctrl = { .action = HOST_DO_NONE, .hostname = "host1" };
+    int rc = xce_hostctrl_req((endpoint_t*)0x1, &ctrl);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+}
+
+static void test_hostctrl_req_reject_tag(void)
+{
+    xce_hostctrl_t ctrl = { .action = HOST_TAG, .hostname = "host1" };
+    int rc = xce_hostctrl_req((endpoint_t*)0x1, &ctrl);
+    TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
+}
+
+static void test_hostctrl_req_reject_untag(void)
+{
+    xce_hostctrl_t ctrl = { .action = HOST_UNTAG, .hostname = "host1" };
+    int rc = xce_hostctrl_req((endpoint_t*)0x1, &ctrl);
+    TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
+}
+
+static void test_hostctrl_req_reject_negative_action(void)
+{
+    xce_hostctrl_t ctrl = { .action = -1, .hostname = "host1" };
+    int rc = xce_hostctrl_req((endpoint_t*)0x1, &ctrl);
+    TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
+}
+
+void test_request(void)
+{
+    printf("\n>> host control request validation tests\n\n");
+    RUN_TEST(test_hostctrl_req_null_ep);
+    RUN_TEST(test_hostctrl_req_null_ctrl);
+    RUN_TEST(test_hostctrl_req_null_hostname);
+    RUN_TEST(test_hostctrl_req_empty_hostname);
+    RUN_TEST(test_hostctrl_req_noop);
+    RUN_TEST(test_hostctrl_req_reject_tag);
+    RUN_TEST(test_hostctrl_req_reject_untag);
+    RUN_TEST(test_hostctrl_req_reject_negative_action);
+}

src/lib/xceapi.c

@@ -561,6 +561,32 @@ int xce_queuectrl(xce_ctx_t* ctx, xce_queuectrl_t* ctrl)
     return status;
 }
 
+int xce_hostctrl(xce_ctx_t* ctx, xce_hostctrl_t* ctrl)
+{
+    int rc;
+    int status;
+
+    if (ctx == NULL) {
+        return -EINVAL;
+    }
+
+    if (ctx->state != XCE_CTX_RUN) {
+        return -EINVAL;
+    }
+
+    rc = xce_hostctrl_req(ctx->ep, ctrl);
+    if (rc < 0) {
+        return rc;
+    }
+
+    rc = cbuf_get(&(ctx->wbuf), &status);
+    if (rc < 0) {
+        return rc;
+    }
+
+    return status;
+}
+
 int xce_runtask(xce_ctx_t* ctx, xce_taskrun_t* runinfo)
 {
     int rc;

src/schema/controller.capnp

@@ -515,6 +515,11 @@ struct QueueControl $C.mapname("xce_queuectrl_t") {
   queue      @1: Text $C.mapname("queue");
 }
 
+struct HostControl $C.mapname("xce_hostctrl_t") {
+  action     @0: UInt32 $C.mapname("action");
+  hostname   @1: Text $C.mapname("hostname");
+}
+
 struct RequestReply $C.mapname("xce_reply_t"){
   seq    @0: UInt64;
   opcode @1: OPCode;
@@ -582,6 +587,7 @@ struct Request $C.mapname("xce_request_t"){
     runtaskmsg      @18: RunTask;
     preasonmsg      @19: PendReasonSet;
     clrpreasonmsg   @20: ClearPReason;
+    hostctrlmsg     @21: HostControl;
   }
 }